From ca3de8586e5afea5e15b0baab9af55e4c42858e8 Mon Sep 17 00:00:00 2001 From: Loren Eteval Date: Mon, 17 Aug 2026 05:13:46 +0800 Subject: [PATCH] Add project tests Signed-off-by: Loren Eteval --- tests/README.md | 81 +++++ tests/__init__.py | 20 ++ tests/support.py | 263 +++++++++++++++++ tests/test_controllers.py | 248 ++++++++++++++++ tests/test_external_core.py | 475 ++++++++++++++++++++++++++++++ tests/test_models_and_services.py | 357 ++++++++++++++++++++++ tests/test_plugin_architecture.py | 367 +++++++++++++++++++++++ tests/test_qt_lifetime.py | 388 ++++++++++++++++++++++++ tests/test_qt_stress.py | 214 ++++++++++++++ tests/test_socks_uri.py | 306 +++++++++++++++++++ tests/test_ui_behavior.py | 265 +++++++++++++++++ 11 files changed, 2984 insertions(+) create mode 100644 tests/README.md create mode 100644 tests/__init__.py create mode 100644 tests/support.py create mode 100644 tests/test_controllers.py create mode 100644 tests/test_external_core.py create mode 100644 tests/test_models_and_services.py create mode 100644 tests/test_plugin_architecture.py create mode 100644 tests/test_qt_lifetime.py create mode 100644 tests/test_qt_stress.py create mode 100644 tests/test_socks_uri.py create mode 100644 tests/test_ui_behavior.py diff --git a/tests/README.md b/tests/README.md new file mode 100644 index 0000000..60c8ff7 --- /dev/null +++ b/tests/README.md @@ -0,0 +1,81 @@ +# Furious test suite + +The suite uses Python's built-in `unittest` runner. Qt tests select the +`offscreen` platform before importing PySide6, construct one deliberately small +test `QApplication`, and route `QSettings` to a unique temporary directory. +They do not initialize Furious's singleton IPC server, production repositories, +system proxy, TUN, routing, update network clients, or real proxy cores. + +## Coverage map + +| Area | Principal tests | +| --- | --- | +| 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` | +| 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` | +| 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` | + +The lifecycle tests classify `AppQTransientDialog`, protocol/plugin editors, +routing dialogs, subscription editors, message boxes, QR windows, and TUN +settings dialogs as transient. `TextEditorWindow` is intentionally reusable: it +must survive normal close/show cycles without multiplying actions, and is then +explicitly destroyed by its owner. Main pages/controllers are application +lifetime objects and are tested through isolated service/UI boundaries rather +than by starting the production application runtime. + +## Commands + +From the repository root on 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. + +## Packaged-build smoke procedure + +Packaged/Nuitka builds cannot be safely driven by these in-process `unittest` +fixtures. For an optional release smoke check, use an otherwise disposable test +OS account or VM, redirect all Furious application-data/settings locations to a +temporary directory, and keep system proxy and TUN disabled. Open and close each +transient editor family 50 times, verify one reusable `TextEditorWindow` does +not duplicate actions, and compare live-object diagnostics from an instrumented +build before/after the loop. Do not run this procedure against a production +profile or rely on process-name cleanup; close only the exact packaged process +started for the smoke test. + +## Isolation rules + +- Tests clean up only exact subprocess handles/PIDs and threads they create. +- Tests never search for, signal, or terminate another Furious/core process. +- Persistence tests use temporary INI-backed `QSettings` namespaces. +- Controller tests inject fake runtime managers and patch host-mutation APIs. +- Qt tests use normal close/deferred-delete paths and collect Python cycles only + at diagnostic batch boundaries, never once per UI operation. +- A lifetime failure must be investigated as an ownership defect; increasing + thresholds or forcing production garbage collection is not an acceptable fix. diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..644c1ba --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1,20 @@ +# 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 . + +"""Contain deterministic Furious regression tests.""" + +from __future__ import annotations diff --git a/tests/support.py b/tests/support.py new file mode 100644 index 0000000..ef1ce1f --- /dev/null +++ b/tests/support.py @@ -0,0 +1,263 @@ +# 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 . + +"""Provide isolated Qt, settings, and lifetime helpers for the test suite.""" + +from __future__ import annotations + +from pathlib import Path +from contextlib import contextmanager + +import os +import gc +import time +import uuid +import ctypes +import tempfile +import weakref + +# Select the headless platform before importing any Qt module. This process is +# deliberately independent from any production Furious process and its native +# windows. +os.environ.setdefault('QT_QPA_PLATFORM', 'offscreen') + +from PySide6 import QtCore +from PySide6.QtWidgets import QApplication + +from shiboken6 import isValid + + +class _DisconnectedController: + """Provide the read-only connection state required by theme-aware widgets.""" + + interactionEnabled = True + + @staticmethod + def isConnected() -> bool: + """Return the deterministic disconnected state used by UI tests.""" + return False + + +class TestApplication(QApplication): + """Provide only application attributes required to construct real widgets.""" + + __test__ = False + + def __init__(self): + """Initialize one side-effect-free application for this test process.""" + super().__init__([]) + + self.setApplicationName('Furious Tests') + self.setOrganizationName('Furious Tests') + self.setOrganizationDomain('tests.invalid') + + self.connectionController = _DisconnectedController() + self.routingController = None + self.settingsController = None + self.systemTray = None + self.mainWindow = None + self.logManager = None + self.logPage = None + self.customFontName = '' + self.threadPool = QtCore.QThreadPool(self) + + @staticmethod + def theme() -> str: + """Return a stable theme without consulting host appearance settings.""" + return 'Dark' + + @staticmethod + def usesForcedDarkTheme() -> bool: + """Return whether the test theme represents a user-forced preference.""" + return False + + @staticmethod + def isExiting() -> bool: + """Return whether the application is currently exiting.""" + return False + + @staticmethod + def applyThemePreference(): + """Accept controller callbacks without mutating host appearance.""" + + +_application = None + + +def application() -> QApplication: + """Return the single QApplication-compatible instance for all tests.""" + global _application + + current = QApplication.instance() + + if current is None: + _application = TestApplication() + + current = _application + + return current + + +def processQtEvents(rounds: int = 3): + """Drain regular and deferred-delete events deterministically.""" + app = application() + + for _index in range(max(1, rounds)): + app.sendPostedEvents(None, QtCore.QEvent.Type.DeferredDelete) + app.processEvents(QtCore.QEventLoop.ProcessEventsFlag.AllEvents, 20) + + +def waitFor(predicate, timeout: float = 2.0) -> bool: + """Pump Qt events until *predicate* succeeds or a bounded timeout expires.""" + deadline = time.monotonic() + timeout + + while time.monotonic() < deadline: + processQtEvents(1) + + if predicate(): + return True + + QtCore.QThread.msleep(1) + + processQtEvents() + + return bool(predicate()) + + +def closeTransient(widget) -> weakref.ReferenceType: + """Close one transient through its production path and return a weak ref.""" + reference = weakref.ref(widget) + + widget.close() + + return reference + + +def collectAtBoundary(): + """Collect Python cycles at a diagnostic batch boundary, then drain Qt.""" + processQtEvents() + + gc.collect() + + processQtEvents() + + +def qObjectCount(qobjectType) -> int: + """Count valid Python wrappers of one application-owned QObject type.""" + count = 0 + + for value in gc.get_objects(): + try: + if isinstance(value, qobjectType) and isValid(value): + count += 1 + except (ReferenceError, RuntimeError): + continue + + return count + + +def currentRSS() -> int | None: + """Return current resident memory where a stable platform API is available.""" + if os.name == 'nt': + + class ProcessMemoryCounters(ctypes.Structure): + """Match the Windows PROCESS_MEMORY_COUNTERS structure.""" + + _fields_ = ( + ('cb', ctypes.c_ulong), + ('PageFaultCount', ctypes.c_ulong), + ('PeakWorkingSetSize', ctypes.c_size_t), + ('WorkingSetSize', ctypes.c_size_t), + ('QuotaPeakPagedPoolUsage', ctypes.c_size_t), + ('QuotaPagedPoolUsage', ctypes.c_size_t), + ('QuotaPeakNonPagedPoolUsage', ctypes.c_size_t), + ('QuotaNonPagedPoolUsage', ctypes.c_size_t), + ('PagefileUsage', ctypes.c_size_t), + ('PeakPagefileUsage', ctypes.c_size_t), + ) + + counters = ProcessMemoryCounters() + counters.cb = ctypes.sizeof(counters) + + kernel32 = ctypes.WinDLL('kernel32', use_last_error=True) + kernel32.GetCurrentProcess.argtypes = () + kernel32.GetCurrentProcess.restype = ctypes.c_void_p + + psapi = ctypes.WinDLL('psapi', use_last_error=True) + psapi.GetProcessMemoryInfo.argtypes = ( + ctypes.c_void_p, + ctypes.POINTER(ProcessMemoryCounters), + ctypes.c_ulong, + ) + psapi.GetProcessMemoryInfo.restype = ctypes.c_int + + process = kernel32.GetCurrentProcess() + + if psapi.GetProcessMemoryInfo( + process, + ctypes.byref(counters), + counters.cb, + ): + return int(counters.WorkingSetSize) + + return None + + statm = Path('/proc/self/statm') + + if statm.exists(): + residentPages = int(statm.read_text(encoding='ascii').split()[1]) + + return residentPages * int(os.sysconf('SC_PAGE_SIZE')) + + return None + + +@contextmanager +def isolatedSettings(): + """Route every QSettings read/write to one temporary test namespace.""" + app = application() + + with tempfile.TemporaryDirectory(prefix='furious-tests-settings-') as directory: + oldOrganization, oldApplication = ( + app.organizationName(), + app.applicationName(), + ) + + namespace = uuid.uuid4().hex + + QtCore.QSettings.setDefaultFormat(QtCore.QSettings.Format.IniFormat) + QtCore.QSettings.setPath( + QtCore.QSettings.Format.IniFormat, + QtCore.QSettings.Scope.UserScope, + directory, + ) + + app.setOrganizationName(f'Furious Tests {namespace}') + app.setApplicationName(f'Furious Tests {namespace}') + + settings = QtCore.QSettings() + settings.clear() + settings.sync() + + try: + yield settings + finally: + settings.clear() + settings.sync() + + app.setOrganizationName(oldOrganization) + app.setApplicationName(oldApplication) diff --git a/tests/test_controllers.py b/tests/test_controllers.py new file mode 100644 index 0000000..9391add --- /dev/null +++ b/tests/test_controllers.py @@ -0,0 +1,248 @@ +# 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 controller state transitions without real networking side effects.""" + +from __future__ import annotations + +from Furious.Controllers.ConnectionController import ( + ConnectionController, + ConnectionState, +) +from Furious.Frozenlib import AppBinarySettings, AppSettings +from Furious.Models import ConfigFactory, ServerProfile +from Furious.Service.LogManager import LogManager + +from tests.support import application, isolatedSettings, processQtEvents + +import unittest + +from unittest import mock + + +class ControllerConfiguration(ConfigFactory): + """Provide a valid local proxy endpoint for controller-only tests.""" + + def httpProxy(self) -> str: + """Return a deterministic loopback proxy endpoint.""" + return '127.0.0.1:18080' + + def coreName(self) -> str: + """Return a deterministic core display name.""" + return 'Fixture Core' + + +class FixtureCoreManager: + """Record lifecycle calls without launching a subprocess or changing routes.""" + + def __init__(self, *, startResult=True, startError=''): + """Initialize deterministic start behavior.""" + self.startResult = startResult + self.lastStartError = startError + self.processesPool = [] + self.startCalls = [] + self.stopCalls = 0 + + def start(self, configuration, **kwargs): + """Record one requested start and return the configured result.""" + self.startCalls.append((configuration, kwargs)) + + return self.startResult + + def stopAll(self): + """Record one bounded cleanup operation.""" + self.stopCalls += 1 + self.processesPool.clear() + + +class FixtureUpdatesManager: + """Record update hooks without contacting any update service.""" + + def __init__(self): + """Initialize empty call history.""" + self.proxy = None + self.checks = 0 + + def configureHttpProxy(self, proxy): + """Record the proxy endpoint passed by post-connect maintenance.""" + self.proxy = proxy + + def checkForUpdates(self, **kwargs): + """Record a suppressed network update check.""" + self.checks += 1 + + +class ConnectionControllerTest(unittest.TestCase): + """Verify state and cleanup while all host mutations are patched out.""" + + def setUp(self): + """Install a fresh structured log manager on the test application.""" + self.app = application() + self.app.logManager = LogManager(parent=self.app) + self.profile = ServerProfile.fromConfiguration( + ControllerConfiguration({'type': 'controller-fixture'}) + ) + + def tearDown(self): + """Release the test-owned log manager.""" + manager = self.app.logManager + + self.app.logManager = None + + if manager is not None: + manager.deleteLater() + + processQtEvents() + + def testSuccessfulConnectionAndDisconnectionStateMachine(self): + """Publish stable states while using only injected runtime resources.""" + with isolatedSettings(): + core = FixtureCoreManager() + controller = ConnectionController( + coreManager=core, + updatesManager=FixtureUpdatesManager(), + ) + states = [] + interactions = [] + controller.stateChanged.connect(states.append) + controller.interactionEnabledChanged.connect(interactions.append) + + with ( + mock.patch( + 'Furious.Controllers.ConnectionController.SystemProxy.set' + ) as proxySet, + mock.patch( + 'Furious.Controllers.ConnectionController.SystemProxy.off' + ) as proxyOff, + mock.patch.object(controller, '_runPostConnectTasksOnce'), + ): + self.assertTrue(controller.startConnection(self.profile)) + self.assertEqual(controller.state, ConnectionState.Connected) + self.assertIs(controller.activeConfiguration, self.profile) + self.assertEqual( + AppSettings.get('Connect'), + AppBinarySettings.ON_, + ) + + proxySet.assert_called_once() + + self.assertTrue(controller.startDisconnection('Stopped')) + + proxyOff.assert_called_once() + + self.assertEqual(controller.state, ConnectionState.Disconnected) + self.assertIsNone(controller.activeConfiguration) + self.assertEqual(core.stopCalls, 1) + self.assertEqual( + states, + [ + ConnectionState.Connecting, + ConnectionState.Connected, + ConnectionState.Disconnecting, + ConnectionState.Disconnected, + ], + ) + self.assertEqual(interactions, [False, True, False, True]) + + controller.deleteLater() + + def testFailedRuntimeStartReturnsToDisconnectedWithError(self): + """Stop a failed launch and expose one user-facing error object.""" + with isolatedSettings(): + core = FixtureCoreManager( + startResult=False, + startError='fixture launch failure', + ) + controller = ConnectionController( + coreManager=core, + updatesManager=FixtureUpdatesManager(), + ) + errors = [] + controller.errorOccurred.connect(errors.append) + + with ( + 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.assertEqual(core.stopCalls, 1) + self.assertIsNone(controller.activeConfiguration) + self.assertEqual( + AppSettings.get('Connect'), + AppBinarySettings.OFF, + ) + # Runtime start failures are currently presented through the + # disconnection notification path, not the preflight error signal. + self.assertEqual(errors, []) + + controller.deleteLater() + + def testInvalidProfileNeverCallsRuntimeOrSystemProxy(self): + """Reject invalid input before any process or host mutation is attempted.""" + with isolatedSettings(): + core = FixtureCoreManager() + controller = ConnectionController( + coreManager=core, + updatesManager=FixtureUpdatesManager(), + ) + + with mock.patch( + 'Furious.Controllers.ConnectionController.SystemProxy.set' + ) as proxySet: + self.assertFalse(controller.startConnection(ConfigFactory())) + + self.assertEqual(core.startCalls, []) + + proxySet.assert_not_called() + + self.assertIsNotNone(controller.lastError) + self.assertEqual(controller.state, ConnectionState.Disconnected) + + controller.deleteLater() + + def testShutdownPreservesReconnectPreference(self): + """Stop only the injected runtime while preserving next-start intent.""" + 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)) + + controller.shutdown() + + self.assertEqual(controller.state, ConnectionState.Disconnected) + self.assertEqual( + AppSettings.get('Connect'), + AppBinarySettings.ON_, + ) + self.assertEqual(core.stopCalls, 1) + + controller.deleteLater() + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_external_core.py b/tests/test_external_core.py new file mode 100644 index 0000000..fbdaa9b --- /dev/null +++ b/tests/test_external_core.py @@ -0,0 +1,475 @@ +# 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 the managed External Core process without third-party executables.""" + +from __future__ import annotations + +from Furious.Backends.ExternalCore import ConfigExternalCore, ExternalCoreProcess +from Furious.Backends.ExternalCore.Plugin import ExternalCorePlugin +from Furious.Plugins.API import SubscriptionItem, SubscriptionResult +from Furious.Plugins.Registry import PluginRegistry +from Furious.Service.ConnectionManager import ConnectionManager +from Furious.Service.DnsResolver import DnsResolver +from Furious.Service.SubscriptionImporter import ( + SubscriptionImportService, + SubscriptionSource, +) + +import os +import sys +import json +import time +import tempfile +import threading +import unittest + +from unittest import mock + +from pathlib import Path + + +class ExternalCoreProcessTest(unittest.TestCase): + """Exercise structured launch, output, failure, and repeated shutdown.""" + + @staticmethod + def waitFor(predicate, timeout: float = 5.0) -> bool: + """Wait for a deterministic fixture condition without a Qt event loop.""" + deadline = time.monotonic() + timeout + + while time.monotonic() < deadline: + if predicate(): + return True + + time.sleep(0.02) + + return predicate() + + @staticmethod + def configuration(arguments, cwd: str, environment=None): + """Build one valid Python-backed External Core fixture profile.""" + return ConfigExternalCore( + { + 'type': 'external-core', + 'executable': str(Path(sys.executable).resolve()), + 'workingDirectory': cwd, + 'arguments': list(arguments), + 'environment': dict(environment or {}), + 'httpProxy': '127.0.0.1:10809', + 'socksProxy': '127.0.0.1:10808', + 'shutdownTimeout': 1, + } + ) + + def testStructuredArgumentsCwdEnvironmentAndOutput(self): + """Pass spaces literally and capture both output streams safely.""" + with tempfile.TemporaryDirectory( + prefix='furious external core ', dir=Path.cwd() + ) as directory: + resultPath = Path(directory) / 'result with spaces.json' + payload = 'argument with spaces' + code = ( + 'import json,os,pathlib,sys,time; ' + 'pathlib.Path(sys.argv[1]).write_text(' + 'json.dumps({"cwd":os.getcwd(),"arg":sys.argv[2],' + '"env":os.environ.get("FURIOUS_EXTERNAL_TEST")}),' + 'encoding="utf-8"); ' + 'print("stdout fixture",flush=True); ' + 'print("stderr fixture",file=sys.stderr,flush=True); ' + 'time.sleep(60)' + ) + messages = [] + runtime = ExternalCoreProcess(msgCallback=messages.append) + config = self.configuration( + ['-u', '-c', code, str(resultPath), payload], + directory, + {'FURIOUS_EXTERNAL_TEST': 'Unicode ✓'}, + ) + + self.assertTrue(runtime.start(config)) + self.assertTrue(self.waitFor(resultPath.exists)) + self.assertTrue( + self.waitFor( + lambda: any('stdout fixture' in message for message in messages) + and any( + '[stderr] stderr fixture' in message for message in messages + ) + ) + ) + + result = json.loads(resultPath.read_text(encoding='utf-8')) + + self.assertEqual(Path(result['cwd']), Path(directory)) + self.assertEqual(result['arg'], payload) + self.assertEqual(result['env'], 'Unicode ✓') + + runtime.stop() + + self.assertFalse(runtime.isAlive()) + self.assertFalse(runtime._readerThreads) + + def testMissingExecutableAndImmediateExitFailStartup(self): + """Report authoritative path and early non-zero-exit failures.""" + with tempfile.TemporaryDirectory(dir=Path.cwd()) as directory: + missing = self.configuration([], directory) + missing['executable'] = str(Path(directory) / 'missing executable') + + runtime = ExternalCoreProcess() + + self.assertFalse(runtime.start(missing)) + self.assertEqual(runtime.startError(), 'Executable does not exist') + + invalidCwd = self.configuration([], directory) + invalidCwd['workingDirectory'] = str(Path(directory) / 'missing cwd') + + self.assertFalse(runtime.start(invalidCwd)) + self.assertEqual( + runtime.startError(), + 'Working directory does not exist', + ) + + invalidEnvironment = self.configuration([], directory) + invalidEnvironment['environment'] = ['TOKEN=value'] + + self.assertFalse(runtime.start(invalidEnvironment)) + self.assertEqual( + runtime.startError(), + 'Environment overrides must be a mapping', + ) + + earlyExit = self.configuration( + ['-c', 'import sys; sys.exit(7)'], + directory, + ) + + self.assertFalse(runtime.start(earlyExit)) + self.assertEqual(runtime.lastExitCode, 7) + self.assertEqual( + runtime.startError(), + 'External core exited during startup', + ) + + runtime.dispose() + + def testApplicationTun2socksUsesOnlyTheConfiguredRemoteAddress(self): + """Keep process paths separate from opt-in TUN routing metadata.""" + with tempfile.TemporaryDirectory(dir=Path.cwd()) as directory: + config = self.configuration([], directory) + executable = config['executable'] + + registry = PluginRegistry() + registry.register(ExternalCorePlugin()) + + self.assertFalse(config.usesApplicationTun2socks()) + self.assertFalse(registry.usesApplicationTun2socks(config)) + self.assertEqual(config.itemAddress, '') + self.assertNotEqual(config.itemAddress, executable) + + config['useApplicationTun2socks'] = True + missingAddressError = ( + 'TUN remote address is required when application ' + 'tun2socks is enabled' + ) + + self.assertIn(missingAddressError, config.validateProcess()) + + for address in ( + 'actual-server.example.com', + '203.0.113.42', + '2001:db8::42', + ): + config['tunRemoteAddress'] = address + + self.assertTrue(config.usesApplicationTun2socks()) + self.assertTrue(registry.usesApplicationTun2socks(config)) + self.assertEqual(config.itemAddress, address) + self.assertNotIn(missingAddressError, config.validateProcess()) + + registry.shutdown() + + def testDisabledApplicationTun2socksSkipsTheHostTunRuntime(self): + """Do not enter ConnectionManager's TUN path for an opted-out profile.""" + + class NoKernelConnectionManager(ConnectionManager): + """Pretend the external process started without launching a child.""" + + def _startKernel(self, *args, **kwargs): + """Return one successful process-free fixture launch.""" + return None, True + + with tempfile.TemporaryDirectory(dir=Path.cwd()) as directory: + config = self.configuration([], directory) + + registry = mock.Mock() + registry.prepareTUN.return_value = False + registry.usesApplicationTun2socks.return_value = False + + manager = NoKernelConnectionManager() + + with ( + mock.patch( + 'Furious.Service.ConnectionManager.SystemRuntime.isTUNMode', + return_value=True, + ), + mock.patch( + 'Furious.Service.ConnectionManager.getPluginRegistry', + return_value=registry, + ), + mock.patch('Furious.Service.ConnectionManager.Tun2socks') as tun2socks, + ): + self.assertTrue(manager.start(config, '', deepcopy=False)) + + registry.usesApplicationTun2socks.assert_called_once_with(config) + tun2socks.assert_not_called() + manager.cleanup() + + def testEnabledApplicationTun2socksResolvesOnlyTheRemoteAddress(self): + """Send the configured network destination, never the executable, to DNS.""" + + class NoKernelConnectionManager(ConnectionManager): + """Pretend the external process started without launching a child.""" + + def _startKernel(self, *args, **kwargs): + """Return one successful process-free fixture launch.""" + return None, True + + with tempfile.TemporaryDirectory(dir=Path.cwd()) as directory: + config = self.configuration([], directory) + config['useApplicationTun2socks'] = True + config['tunRemoteAddress'] = 'actual-server.example.com' + + executable = config['executable'] + + registry = mock.Mock() + registry.prepareTUN.return_value = False + registry.usesApplicationTun2socks.return_value = True + + manager = NoKernelConnectionManager() + + with ( + mock.patch( + 'Furious.Service.ConnectionManager.SystemRuntime.isTUNMode', + return_value=True, + ), + mock.patch( + 'Furious.Service.ConnectionManager.getPluginRegistry', + return_value=registry, + ), + mock.patch( + 'Furious.Service.ConnectionManager.userDefaultPrimaryGatewayIP', + return_value='192.168.50.1', + ), + mock.patch( + 'Furious.Service.ConnectionManager.userPrimaryAdapterInterfaceIP', + return_value='192.168.50.20', + ), + mock.patch( + 'Furious.Service.ConnectionManager.userTcpSendBufferSize', + return_value=1, + ), + mock.patch( + 'Furious.Service.ConnectionManager.userTcpReceiveBufferSize', + return_value=1, + ), + mock.patch( + 'Furious.Service.ConnectionManager.userTcpAutoTuning', + return_value='False', + ), + mock.patch( + 'Furious.Service.ConnectionManager.userBypassTUNAdapterInterfaceIP', + return_value='', + ), + mock.patch( + 'Furious.Service.ConnectionManager.SystemRoutingTable.delete' + ), + mock.patch('Furious.Service.ConnectionManager.Tun2socks'), + mock.patch( + 'Furious.Service.ConnectionManager.DnsResolver.configureHttpProxy' + ), + mock.patch( + 'Furious.Service.ConnectionManager.DnsResolver.resolve', + return_value=(True, []), + ) as resolve, + ): + self.assertFalse(manager.start(config, '', deepcopy=False)) + + resolve.assert_called_once_with('actual-server.example.com') + + self.assertNotEqual(resolve.call_args.args[0], executable) + + manager.cleanup() + + def testMissingTunRemoteAddressFailsBeforeProcessLaunch(self): + """Reject opted-in TUN integration before spawning the executable.""" + with tempfile.TemporaryDirectory(dir=Path.cwd()) as directory: + config = self.configuration([], directory) + config['useApplicationTun2socks'] = True + + runtime = ExternalCoreProcess() + + self.assertFalse(runtime.start(config)) + self.assertIsNone(runtime.process) + self.assertEqual( + runtime.startError(), + 'TUN remote address is required when application ' + 'tun2socks is enabled', + ) + + runtime.dispose() + + def testUnexpectedExitCallsBackAndRepeatedShutdownDoesNotRetainThreads(self): + """Notice post-start failure and leave no workers across restarts.""" + with tempfile.TemporaryDirectory(dir=Path.cwd()) as directory: + callbackEvent = threading.Event() + callbackValues = [] + + def exited(runtime, exitCode): + """Record one unexpected process exit from the watcher thread.""" + callbackValues.append((runtime, exitCode)) + callbackEvent.set() + + runtime = ExternalCoreProcess(exitCallback=exited) + config = self.configuration( + ['-c', 'import time,sys; time.sleep(.5); sys.exit(9)'], + directory, + ) + + self.assertTrue(runtime.start(config)) + self.assertTrue(callbackEvent.wait(5)) + self.assertEqual(callbackValues, [(runtime, 9)]) + + runtime.stop() + + longRunning = self.configuration( + ['-c', 'import time; time.sleep(60)'], + directory, + ) + + for _index in range(3): + self.assertTrue(runtime.start(longRunning)) + + runtime.stop() + + self.assertFalse(runtime.isAlive()) + self.assertFalse(runtime._readerThreads) + self.assertIsNone(runtime._watcherThread) + + runtime.dispose() + + @unittest.skipUnless(os.name == 'nt', 'Windows executable hard-link coverage') + def testExecutablePathContainingSpaces(self): + """Launch an executable hard link whose local path contains spaces.""" + with tempfile.TemporaryDirectory( + prefix='furious executable path ', dir=Path.cwd() + ) as directory: + executable = Path(directory) / 'python executable.exe' + + os.link(sys.executable, executable) + + config = self.configuration( + ['-c', 'import time; time.sleep(60)'], + directory, + ) + config['executable'] = str(executable) + + runtime = ExternalCoreProcess() + + self.assertTrue(runtime.start(config)) + + runtime.stop() + + self.assertFalse(runtime.isAlive()) + + def testSubscriptionCannotIntroduceAnExecutableProfile(self): + """Reject executable configurations received from subscription data.""" + + class ExternalConfigurationRegistry(PluginRegistry): + """Return one deterministic untrusted subscription item.""" + + def decodeSubscription(self, data: bytes, decoderId=None): + """Return an External Core mapping regardless of payload.""" + return SubscriptionResult( + 'fixture', + ( + SubscriptionItem( + configuration={ + 'type': 'external-core', + 'executable': str(Path(sys.executable).resolve()), + 'arguments': [], + 'environment': {}, + 'httpProxy': '127.0.0.1:10809', + } + ), + ), + ) + + registry = ExternalConfigurationRegistry() + registry.register(ExternalCorePlugin()) + + result = SubscriptionImportService(registry).importPayload( + b'untrusted', + SubscriptionSource('fixture'), + ) + + self.assertEqual(result.profiles, tuple()) + self.assertEqual(result.rejectedItems, 1) + + registry.shutdown() + + +class DnsResolverRobustnessTest(unittest.TestCase): + """Verify expected negative DNS responses do not raise raw exceptions.""" + + def testResponseWithoutAnswerReportsResolutionFailure(self): + """Treat a valid DNS JSON response without Answer as a normal failure.""" + + class ReplyData: + """Provide the QByteArray-compatible method used by the resolver.""" + + @staticmethod + def data(): + """Return a deterministic NXDOMAIN-style DNS response.""" + return b'{"Status":3,"Comment":"NXDOMAIN"}' + + class Reply: + """Return the fixture response body through the network-reply API.""" + + @staticmethod + def readAll(): + """Return the response data wrapper.""" + return ReplyData() + + result = { + 'error': False, + 'depth': 1, + 'reference': [], + 'result': {}, + } + + DnsResolver.successCallback( + Reply(), + domain='missing.example', + resultMap=result, + ) + + self.assertTrue(result['error']) + self.assertEqual(result['depth'], 0) + self.assertEqual(result['result'], {}) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_models_and_services.py b/tests/test_models_and_services.py new file mode 100644 index 0000000..d99bdd2 --- /dev/null +++ b/tests/test_models_and_services.py @@ -0,0 +1,357 @@ +# 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 models, persistence, logging, metrics, and settings migration.""" + +from __future__ import annotations + +from Furious.Controllers.SettingsController import ( + APPLICATION_THEME_SETTING, + SettingsController, +) +from Furious.Frozenlib import AppBinarySettings, AppSettings, ApplicationTheme +from Furious.Models import ConfigFactory, ProfileMetadata, ServerProfile +from Furious.Repository.Servers import UserServer, UserServers +from Furious.Repository.Subscriptions import SubscriptionGroup, UserSubs +from Furious.Service.LogManager import ( + APPLICATION_LOG_CATEGORY, + CORE_LOG_CATEGORY, + LogManager, +) +from Furious.Service.MetricsDataManager import ( + DOWNLOAD_SPEED_METRIC, + DOWNLOAD_USAGE_METRIC, + MetricsDataManager, +) + +from PySide6 import QtCore + +import threading +import unittest + +from tests.support import isolatedSettings + + +class ProfileModelTest(unittest.TestCase): + """Verify metadata separation, compatibility, and copy semantics.""" + + def testLegacyMetadataPreservesUnknownFields(self): + """Promote known legacy fields while retaining forward-only metadata.""" + metadata = ProfileMetadata.fromMapping( + { + 'remark': 'Legacy name', + 'subsId': 'subscription-id', + 'delayResult': '42 ms', + 'speedResult': '8 MiB/s', + 'tags': 'work, ipv6', + 'favorite': 'true', + 'futureMetadata': {'value': 7}, + } + ) + + self.assertEqual(metadata.displayName, 'Legacy name') + self.assertEqual(metadata.subscriptionSource, 'subscription-id') + self.assertTrue(metadata.subscriptionManaged) + self.assertEqual(metadata.latency, '42 ms') + self.assertEqual(metadata.speed, '8 MiB/s') + self.assertEqual(metadata.tags, ('work', 'ipv6')) + self.assertTrue(metadata.favorite) + self.assertEqual(metadata.extras['futureMetadata'], {'value': 7}) + self.assertEqual( + ProfileMetadata.fromMapping(metadata.toMapping()).toMapping(), + metadata.toMapping(), + ) + + def testIndependentCopyGetsNewIdentityAndNoSubscriptionOwner(self): + """Keep manual copies independent from subscription synchronization.""" + original = ServerProfile.fromConfiguration( + ConfigFactory({'type': 'fixture', 'address': 'example.com'}), + { + 'displayName': 'Managed', + 'subscriptionSource': 'source', + 'subscriptionManaged': True, + 'subscriptionProfileKey': 'upstream:1', + }, + ) + copied = original.independentCopy() + + self.assertNotEqual(copied.metadata.profileId, original.metadata.profileId) + self.assertEqual(copied.connection, original.connection) + self.assertIsNot(copied.connection, original.connection) + self.assertEqual(copied.metadata.subscriptionSource, '') + self.assertFalse(copied.metadata.subscriptionManaged) + self.assertEqual(copied.metadata.subscriptionProfileKey, '') + + def testUserServerMappingRemainsBackwardCompatible(self): + """Persist the canonical legacy record shape plus per-profile metadata.""" + profile = ServerProfile.fromConfiguration( + ConfigFactory({'type': 'fixture', 'port': 1080}), + { + 'displayName': 'Fixture', + 'group': 'Tests', + 'annotations': 'local only', + 'favorite': True, + 'futureField': 'preserved', + }, + ) + mapping = UserServer.fromProfile(profile).toMapping() + restored = UserServer.metadataFromMapping(mapping) + + self.assertEqual(set(('remark', 'config', 'subsId')) - set(mapping), set()) + self.assertEqual(restored.displayName, 'Fixture') + self.assertEqual(restored.group, 'Tests') + self.assertEqual(restored.annotations, 'local only') + self.assertTrue(restored.favorite) + self.assertEqual(restored.extras['futureField'], 'preserved') + + +class IsolatedRepositoryTest(unittest.TestCase): + """Prove repositories round-trip exclusively through temporary QSettings.""" + + def testServerRepositoryRoundTripUsesCanonicalModel(self): + """Restore connection and metadata without touching production state.""" + with isolatedSettings() as settings: + repository = UserServers() + repository.data().append( + ServerProfile.fromConfiguration( + ConfigFactory({'type': 'fixture', 'value': 9}), + { + 'displayName': 'Temporary profile', + 'tags': ('one', 'two'), + 'favorite': True, + 'unknown': 'retained', + }, + ) + ) + repository.sync() + + self.assertTrue(settings.contains('Configuration')) + + restored = UserServers().data() + + self.assertEqual(len(restored), 1) + self.assertEqual(restored[0].connection['value'], 9) + self.assertEqual(restored[0].itemRemark, 'Temporary profile') + self.assertEqual(restored[0].metadata.tags, ('one', 'two')) + self.assertTrue(restored[0].metadata.favorite) + self.assertEqual(restored[0].metadata.extras['unknown'], 'retained') + + def testSubscriptionRepositoryNormalizesAndOrdersLegacyGroups(self): + """Keep future fields while migrating URL-era subscription records.""" + with isolatedSettings(): + repository = UserSubs() + repository.upsertGroup( + SubscriptionGroup.fromMapping( + 'second', + { + 'remark': 'Zulu', + 'webURL': 'https://two.invalid', + 'sortOrder': 2, + 'futureField': 'two', + }, + ) + ) + repository.upsertGroup( + SubscriptionGroup.fromMapping( + 'first', + { + 'remark': 'Alpha', + 'webURL': 'https://one.invalid', + 'sortOrder': 1, + 'enabled': 'false', + }, + ) + ) + repository.sync() + + restored = UserSubs() + groups = restored.groups() + + self.assertEqual(tuple(group.id for group in groups), ('first', 'second')) + self.assertFalse(groups[0].enabled) + self.assertEqual(groups[1].extras['futureField'], 'two') + self.assertEqual(restored.removeGroup('first').remark, 'Alpha') + self.assertIsNone(restored.group('first')) + + +class SettingsMigrationTest(unittest.TestCase): + """Verify forward/backward-compatible application-theme persistence.""" + + def testLegacyDarkModeMigratesWithoutRemovingLegacyValue(self): + """Create the new preference while retaining the old binary key.""" + with isolatedSettings() as settings: + settings.setValue('DarkMode', AppBinarySettings.ON_) + + SettingsController() + + self.assertEqual( + AppSettings.get(APPLICATION_THEME_SETTING), + ApplicationTheme.Dark.value, + ) + self.assertEqual(settings.value('DarkMode'), AppBinarySettings.ON_) + + def testNewPreferenceSynchronizesLegacyReaders(self): + """Keep older releases able to read forced dark and non-dark choices.""" + with isolatedSettings() as settings: + SettingsController() + + SettingsController.setApplicationTheme(ApplicationTheme.Dark) + + self.assertEqual(settings.value('DarkMode'), AppBinarySettings.ON_) + + SettingsController.setApplicationTheme(ApplicationTheme.Light) + + self.assertEqual(settings.value('DarkMode'), AppBinarySettings.OFF) + self.assertEqual( + settings.value(APPLICATION_THEME_SETTING), + ApplicationTheme.Light.value, + ) + + +class LogManagerTest(unittest.TestCase): + """Verify bounded, categorized, and thread-safe structured logging.""" + + def testBoundedBufferCategoriesAndRuntimeClear(self): + """Retain only the newest entries and clear runtime categories alone.""" + manager = LogManager(maximumEntries=4) + manager.append('application 1') + manager.append('core 1', CORE_LOG_CATEGORY) + manager.append('application 2') + manager.append('core 2', CORE_LOG_CATEGORY) + manager.append('application 3') + + sequence, entries = manager.snapshot() + + self.assertEqual(sequence, 5) + self.assertEqual( + tuple(entry.message for entry in entries), + ('core 1', 'application 2', 'core 2', 'application 3'), + ) + self.assertEqual( + tuple(entry.message for entry in manager.entries(CORE_LOG_CATEGORY)), + ('core 1', 'core 2'), + ) + + manager.clear(runtimeOnly=True) + + self.assertEqual( + tuple(entry.categoryId for entry in manager.entries()), + (APPLICATION_LOG_CATEGORY, APPLICATION_LOG_CATEGORY), + ) + + def testConcurrentProducersReceiveUniqueOrderedSequences(self): + """Serialize worker-thread appends without losing or duplicating entries.""" + manager = LogManager(maximumEntries=500) + + def produce(prefix): + """Append one deterministic worker batch.""" + for index in range(100): + manager.append(f'{prefix}-{index}') + + workers = tuple( + threading.Thread(target=produce, args=(prefix,)) + for prefix in ('a', 'b', 'c') + ) + + for worker in workers: + worker.start() + for worker in workers: + worker.join(5) + + entries = manager.entries() + sequences = tuple(entry.sequence for entry in entries) + + self.assertTrue(all(not worker.is_alive() for worker in workers)) + self.assertEqual(len(entries), 300) + self.assertEqual(sequences, tuple(range(1, 301))) + self.assertEqual(len({entry.message for entry in entries}), 300) + + +class MetricsDataManagerTest(unittest.TestCase): + """Verify bounded history and metric-specific aggregation semantics.""" + + def testPruningNormalizationAndAggregation(self): + """Prune stale values, average speed, and retain latest usage.""" + manager = MetricsDataManager(maximumHistorySeconds=10) + changed = [] + manager.historyChanged.connect(lambda: changed.append(True)) + + manager.recordSample( + { + DOWNLOAD_SPEED_METRIC: -5, + DOWNLOAD_USAGE_METRIC: 100, + 'unknown': 9, + }, + sampledAt=0, + ) + manager.recordSample( + { + DOWNLOAD_SPEED_METRIC: 20, + DOWNLOAD_USAGE_METRIC: 150, + }, + sampledAt=11, + ) + manager.recordSample( + { + DOWNLOAD_SPEED_METRIC: 40, + DOWNLOAD_USAGE_METRIC: 190, + }, + sampledAt=19, + ) + + self.assertEqual(manager.sampleCount(), 2) + self.assertEqual(len(changed), 3) + self.assertEqual( + tuple( + point.value + for point in manager.series( + DOWNLOAD_SPEED_METRIC, + 20, + granularitySeconds=20, + now=19, + ) + ), + (30.0,), + ) + self.assertEqual( + tuple( + point.value + for point in manager.series( + DOWNLOAD_USAGE_METRIC, + 20, + granularitySeconds=20, + now=21, + ) + ), + (190.0,), + ) + + def testInvalidSamplesDoNotPolluteHistory(self): + """Ignore unsupported and non-finite values without emitting changes.""" + manager = MetricsDataManager() + changed = [] + manager.historyChanged.connect(lambda: changed.append(True)) + + manager.recordSample({'unknown': 1}, sampledAt=1) + manager.recordSample({DOWNLOAD_SPEED_METRIC: float('nan')}, sampledAt=2) + + self.assertEqual(manager.rawSamples(), tuple()) + self.assertEqual(changed, []) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_plugin_architecture.py b/tests/test_plugin_architecture.py new file mode 100644 index 0000000..748c1bf --- /dev/null +++ b/tests/test_plugin_architecture.py @@ -0,0 +1,367 @@ +# 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 plugin capability discovery, factories, rollback, and shutdown.""" + +from __future__ import annotations + +from Furious.Models import ConfigFactory +from Furious.Plugins.API import ( + CapabilityKind, + FuriousPlugin, + KernelFactory, + KernelLaunch, + KernelRequest, + PluginMetadata, + ProtocolDescriptor, + ProtocolEditorProvider, + ProtocolHandler, + ProtocolParseResult, + SubscriptionDecoder, + SubscriptionItem, + SubscriptionResult, +) +from Furious.Plugins.Registry import PluginRegistry +from Furious.Service.SubscriptionImporter import ( + SubscriptionImportService, + SubscriptionSource, +) + +import unittest + + +class FixtureConfiguration(ConfigFactory): + """Represent one deterministic plugin-owned connection document.""" + + @property + def itemProtocol(self): + """Return the protocol ID used by display and dispatch code.""" + return 'FIXTURE' + + def httpProxy(self) -> str: + """Return the local HTTP endpoint used by controller fixtures.""" + return '127.0.0.1:18080' + + +class FixtureProtocolHandler(ProtocolHandler): + """Parse and export a minimal fixture:// URI.""" + + descriptor = ProtocolDescriptor( + 'FIXTURE', + 'Fixture', + 'Add Fixture...', + subscriptionImportable=True, + ) + schemes = ('fixture',) + + def supports(self, configuration) -> bool: + """Return whether this handler owns *configuration*.""" + return isinstance( + getattr(configuration, 'connection', configuration), + FixtureConfiguration, + ) + + def parse(self, uri: str, **kwargs): + """Parse one fixture endpoint or decline another scheme.""" + if not uri.casefold().startswith('fixture://'): + return None + + value = uri.split('://', 1)[1] + + if not value: + return None + + return ProtocolParseResult( + FixtureConfiguration({'type': 'fixture', 'value': value}), + {'displayName': value}, + ) + + def fromMapping(self, configuration, **kwargs): + """Recognize normalized fixture mappings.""" + if configuration.get('type') == 'fixture': + return FixtureConfiguration(dict(configuration)) + + return None + + def blank(self, **kwargs): + """Create one valid blank fixture configuration.""" + return FixtureConfiguration({'type': 'fixture', 'value': 'blank'}) + + def export(self, configuration, remark: str = '') -> str: + """Export a fixture URI.""" + connection = getattr(configuration, 'connection', configuration) + + return f"fixture://{connection.get('value', '')}" + + def validate(self, configuration): + """Require a non-empty fixture value.""" + connection = getattr(configuration, 'connection', configuration) + + return tuple() if connection.get('value') else ('missing fixture value',) + + +class FixtureEditorProvider(ProtocolEditorProvider): + """Create a sentinel editor without retaining it in the registry.""" + + editorId = 'fixture.editor' + protocolIds = ('FIXTURE',) + + def createEditor(self, protocolId: str, parent=None, **kwargs): + """Return a fresh sentinel editor for the exact protocol.""" + return {'protocol': protocolId, 'parent': parent, **kwargs} + + +class FixtureKernel: + """Record start calls made through KernelLaunch.""" + + def __init__(self): + """Initialize an empty call history.""" + self.calls = [] + + def start(self, configuration, *args, **kwargs): + """Record prepared launch values and report success.""" + self.calls.append((configuration, args, kwargs)) + + return True + + +class FixtureKernelFactory(KernelFactory): + """Create a deterministic in-process runtime kernel.""" + + factoryId = 'fixture.kernel' + configurationTypes = (FixtureConfiguration,) + kernelTypes = (FixtureKernel,) + + def create(self, request: KernelRequest): + """Build one prepared launch for a fixture configuration.""" + if not isinstance(request.configuration, FixtureConfiguration): + return None + + return KernelLaunch( + FixtureKernel(), + request.configuration, + ('prepared',), + {'routing': request.routing}, + ) + + +class FixtureDecoder(SubscriptionDecoder): + """Decode a deterministic fixture payload.""" + + decoderId = 'fixture-decoder' + displayName = 'Fixture Decoder' + priority = 100 + + def decode(self, data: bytes): + """Decode the exact fixture marker.""" + if data != b'fixture': + return None + + return SubscriptionResult( + self.decoderId, + ( + SubscriptionItem(uri='fixture://one', upstreamId='one'), + SubscriptionItem(uri='fixture://two', upstreamId='two'), + ), + ) + + +class FixturePlugin(FuriousPlugin): + """Bundle representative capabilities for registry tests.""" + + metadata = PluginMetadata('tests.fixture', 'Fixture Plugin') + capabilities = ( + FixtureProtocolHandler(), + FixtureEditorProvider(), + FixtureKernelFactory(), + FixtureDecoder(), + ) + + def __init__(self): + """Initialize lifecycle counters.""" + self.initialized = 0 + self.stopped = 0 + self.context = None + + def initialize(self, context): + """Record one successful initialization.""" + self.initialized += 1 + self.context = context + + def shutdown(self): + """Record one registry-owned shutdown.""" + self.stopped += 1 + + +class PluginRegistryTest(unittest.TestCase): + """Exercise capability dispatch without the process-wide registry.""" + + def setUp(self): + """Create one isolated registry and fixture plugin.""" + self.registry = PluginRegistry() + self.plugin = FixturePlugin() + self.registry.register(self.plugin) + + def tearDown(self): + """Release only plugins owned by this test.""" + self.registry.shutdown() + + def testCapabilityDiscoveryAndMetadata(self): + """Query capabilities without assuming every plugin is a core.""" + self.assertEqual(self.plugin.initialized, 1) + self.assertIs(self.plugin.context.registry, self.registry) + self.assertEqual( + self.registry.metadataFor('tests.fixture'), + self.plugin.metadata, + ) + self.assertEqual( + len(self.registry.capabilities(plugin=self.plugin)), + 4, + ) + self.assertIs( + self.registry.capability(CapabilityKind.Protocol, 'FIXTURE'), + self.plugin.capabilities[0], + ) + self.assertEqual( + self.registry.pluginsWithCapability(CapabilityKind.KernelFactory), + (self.plugin,), + ) + + def testProtocolParseExportValidationAndEditorFactory(self): + """Dispatch URI and editor operations through registered capabilities.""" + parsed = self.registry.parseURI('FiXtUrE://server') + + self.assertIsInstance(parsed.configuration, FixtureConfiguration) + self.assertEqual(parsed.metadata['displayName'], 'server') + self.assertEqual( + self.registry.exportConfig(parsed.configuration), + 'fixture://server', + ) + self.assertEqual(self.registry.validateConfig(parsed.configuration), tuple()) + self.assertEqual( + self.registry.createEditorForConfig(parsed.configuration, marker=7), + {'protocol': 'fixture', 'parent': None, 'marker': 7}, + ) + + def testConfigurationAndKernelFactories(self): + """Build normalized configurations and start one prepared runtime.""" + config = self.registry.configFromDict({'type': 'fixture', 'value': 'node'}) + launch = self.registry.createKernel(config, 'direct') + + self.assertIsInstance(config, FixtureConfiguration) + self.assertIsInstance(launch.kernel, FixtureKernel) + self.assertTrue(launch.start()) + self.assertEqual( + launch.kernel.calls, + [(config, ('prepared',), {'routing': 'direct'})], + ) + + def testSubscriptionImportSeparatesMetadataAndConnection(self): + """Convert decoder items into managed profiles with stable identities.""" + result = SubscriptionImportService(self.registry).importPayload( + b'fixture', + SubscriptionSource('source-id', displayName='Fixture source'), + ) + + self.assertEqual(result.decoderId, 'fixture-decoder') + self.assertEqual(result.rejectedItems, 0) + self.assertEqual(len(result.profiles), 2) + self.assertEqual( + tuple(profile.itemRemark for profile in result.profiles), + ('one', 'two'), + ) + self.assertTrue( + all(profile.metadata.subscriptionManaged for profile in result.profiles) + ) + self.assertEqual( + tuple( + profile.metadata.subscriptionProfileKey for profile in result.profiles + ), + ('upstream:one', 'upstream:two'), + ) + + def testShutdownRunsOnceInReverseRegistryLifetime(self): + """Make shutdown idempotent and reject use of a closed registry.""" + self.registry.shutdown() + self.registry.shutdown() + + self.assertEqual(self.plugin.stopped, 1) + + with self.assertRaises(RuntimeError): + self.registry.register(FixturePlugin()) + + +class PluginRollbackTest(unittest.TestCase): + """Verify failed plugins leave no capabilities or live lifecycle state.""" + + def testInitializationFailureRollsBackEveryIndex(self): + """Remove a failed plugin from all capability lookup structures.""" + + class FailingPlugin(FixturePlugin): + """Raise after capabilities have been indexed.""" + + metadata = PluginMetadata('tests.failure', 'Failing Plugin') + + def initialize(self, context): + """Simulate plugin initialization failure.""" + raise RuntimeError('fixture failure') + + registry = PluginRegistry() + plugin = FailingPlugin() + + with self.assertRaisesRegex(RuntimeError, 'fixture failure'): + registry.register(plugin) + + self.assertEqual(plugin.stopped, 1) + self.assertEqual(registry.plugins(), tuple()) + self.assertEqual(registry.capabilities(), tuple()) + self.assertIsNone(registry.parseURI('fixture://node')) + + registry.shutdown() + + def testDuplicateSchemeDoesNotPartiallyRegisterSecondPlugin(self): + """Reject conflicting protocol ownership atomically.""" + + class ConflictingPlugin(FuriousPlugin): + """Claim an existing URI scheme under another protocol ID.""" + + metadata = PluginMetadata('tests.conflict', 'Conflict') + + class Handler(FixtureProtocolHandler): + """Use a distinct protocol ID with the same scheme.""" + + descriptor = ProtocolDescriptor('OTHER', 'Other', 'Add Other') + + capabilities = (Handler(),) + + first = FixturePlugin() + + registry = PluginRegistry() + registry.register(first) + + try: + with self.assertRaises(ValueError): + registry.register(ConflictingPlugin()) + + self.assertEqual(registry.plugins(), (first,)) + self.assertIsNotNone(registry.parseURI('fixture://node')) + finally: + registry.shutdown() + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_qt_lifetime.py b/tests/test_qt_lifetime.py new file mode 100644 index 0000000..040ada3 --- /dev/null +++ b/tests/test_qt_lifetime.py @@ -0,0 +1,388 @@ +# 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 . + +"""Prove intended Qt ownership across representative transient UI families.""" + +from __future__ import annotations + +from Furious.Backends.ExternalCore.Editor import ExternalCoreEditor +from Furious.Backends.Hysteria2.TunSettingsDialog import Hysteria2TunSettingsDialog +from Furious.Backends.Xray.RoutingWindow import ( + RoutingPreviewDialog, + RoutingRuleEditDialog, + RoutingRulesDialog, +) +from Furious.Backends.Xray.SocksEditor import SocksEditor +from Furious.Backends.Xray.TunSettingsDialog import XrayTunSettingsDialog +from Furious.Frozenlib import Mixins +from Furious.Qt import ( + AppQAction, + AppQDialog, + AppQMainWindow, + AppQMenu, + AppQMessageBox, + AppQTransientDialog, +) +from Furious.Qt.QtWidgets import _AppMessageBoxMask +from Furious.Window.QRCodeWindow import QRCodeWindow +from Furious.Window.SubscriptionPage import _SubscriptionEditorDialog +from Furious.Window.TextEditorWindow import TextEditorWindow + +from PySide6 import QtCore +from PySide6.QtWidgets import QWidget + +from shiboken6 import isValid + +from tests.support import ( + application, + collectAtBoundary, + isolatedSettings, + processQtEvents, + waitFor, +) + +import gc +import unittest +import weakref + + +class ProbeTransientDialog(AppQTransientDialog): + """Own a running timer and menu so their destruction can be observed.""" + + def __init__(self, parent=None): + """Initialize representative transient QObject resources.""" + super().__init__(parent) + + self.timer = QtCore.QTimer(self) + self.timer.start(1000) + + self.action = AppQAction('Fixture action') + self.menu = AppQMenu(self.action, parent=self) + + +class ProbeWindow(AppQMainWindow): + """Expose AppQMainWindow's asynchronous lifetime policy.""" + + +class LongLivedEmitter(QtCore.QObject): + """Represent one application-lifetime signal sender.""" + + emitted = QtCore.Signal() + + +class TransientReceiver(AppQTransientDialog): + """Connect a transient receiver to a long-lived sender.""" + + def __init__(self, emitter, calls): + """Connect exactly one same-process receiver callback.""" + super().__init__() + + self._calls = calls + + emitter.emitted.connect(self.handleEmission) + + @QtCore.Slot() + def handleEmission(self): + """Record one signal delivery.""" + self._calls.append(1) + + +class QtLifetimeTest(unittest.TestCase): + """Stress direct destruction evidence without relying on process RSS alone.""" + + @classmethod + def setUpClass(cls): + """Create the one QApplication used by the entire test process.""" + application() + + def tearDown(self): + """Drain deferred deletion and verify async registries are quiescent.""" + collectAtBoundary() + + self.assertEqual(AppQDialog._openDialogs, {}) + self.assertEqual(AppQMessageBox._openMessageBoxes, {}) + + def assertAllDestroyed(self, references, destroyed, expected): + """Assert weak wrappers and native destroyed signals agree.""" + self.assertTrue( + waitFor(lambda: all(reference() is None for reference in references)), + f'{sum(reference() is not None for reference in references)} wrappers remain', + ) + self.assertEqual(len(destroyed), expected) + + def testTransientDialogTimerMenuAndActionAreDestroyedForEveryCycle(self): + """Destroy timers, menus, actions, and weak-pool registrations 150 times.""" + iterations = 150 + dialogs, timers, menus, actions, destroyed = [], [], [], [], [] + poolBaselines = { + pool: len(pool.ObjectsPool) + for pool in ( + Mixins.ConnectionAware, + Mixins.ThemeAware, + Mixins.QTranslatable, + ) + } + + for _index in range(iterations): + dialog = ProbeTransientDialog() + dialog.destroyed.connect(lambda *_args: destroyed.append(True)) + + dialogs.append(weakref.ref(dialog)) + + timers.append(weakref.ref(dialog.timer)) + menus.append(weakref.ref(dialog.menu)) + actions.append(weakref.ref(dialog.action)) + + dialog.show() + dialog.close() + + del dialog + + collectAtBoundary() + + self.assertAllDestroyed(dialogs, destroyed, iterations) + self.assertTrue(all(reference() is None for reference in timers)) + self.assertTrue(all(reference() is None for reference in menus)) + self.assertTrue(all(reference() is None for reference in actions)) + + for pool, baseline in poolBaselines.items(): + self.assertEqual(len(pool.ObjectsPool), baseline) + + def testAsyncDialogRegistryRetainsOnlyUntilNormalClose(self): + """Prevent premature GC while open and release immediately after finish.""" + dialog = ProbeTransientDialog() + key = dialog._lifetimeKey + reference = weakref.ref(dialog) + dialog.open() + + del dialog + + collectAtBoundary() + + self.assertIsNotNone(reference()) + self.assertIn(key, AppQDialog._openDialogs) + + reference().close() + + collectAtBoundary() + + self.assertTrue(waitFor(lambda: reference() is None)) + self.assertNotIn(key, AppQDialog._openDialogs) + + def testMainWindowRegistryPreventsPrematureCollectionAndReleasesOnClose(self): + """Keep asynchronous top-level windows visible without leaking after close.""" + window = ProbeWindow() + key = window._lifetimeKey + reference = weakref.ref(window) + window.show() + + del window + + collectAtBoundary() + + self.assertIsNotNone(reference()) + self.assertTrue(reference().isVisible()) + self.assertIn(key, AppQMainWindow._openWindows) + + reference().close() + + collectAtBoundary() + + self.assertTrue(waitFor(lambda: reference() is None)) + self.assertNotIn(key, AppQMainWindow._openWindows) + + def testMessageBoxAndParentMaskHaveTransientOwnership(self): + """Remove every parent event filter/mask over repeated modal presentation.""" + iterations = 60 + + owner = QWidget() + owner.resize(640, 480) + owner.show() + + references = [] + maskReferences = [] + destroyed = [] + + for _index in range(iterations): + messageBox = AppQMessageBox( + icon=AppQMessageBox.Icon.Information, + parent=owner, + text='Fixture information', + buttons=AppQMessageBox.StandardButton.Ok, + ) + messageBox.destroyed.connect(lambda *_args: destroyed.append(True)) + messageBox.open() + + processQtEvents() + + references.append(weakref.ref(messageBox)) + maskReferences.append(weakref.ref(messageBox._windowMask)) + messageBox.close() + + del messageBox + + collectAtBoundary() + + self.assertAllDestroyed(references, destroyed, iterations) + self.assertTrue(all(reference() is None for reference in maskReferences)) + self.assertEqual(owner.findChildren(_AppMessageBoxMask), []) + + owner.close() + owner.deleteLater() + + def testLongLivedSenderDoesNotRetainClosedReceiversOrMultiplyCallbacks(self): + """Disconnect deleted receivers and deliver once to the current receiver.""" + emitter = LongLivedEmitter() + calls = [] + oldReferences = [] + + for _index in range(100): + receiver = TransientReceiver(emitter, calls) + + oldReferences.append(weakref.ref(receiver)) + + receiver.show() + receiver.close() + + del receiver + + collectAtBoundary() + + self.assertTrue(all(reference() is None for reference in oldReferences)) + + current = TransientReceiver(emitter, calls) + current.show() + + emitter.emitted.emit() + + processQtEvents() + + self.assertEqual(calls, [1]) + + current.close() + + del current + + emitter.deleteLater() + + def testRepresentativePluginEditorsAndDialogsAreTransient(self): + """Destroy independent editor families after repeated normal closes.""" + factories = ( + ('external-core', ExternalCoreEditor, 35), + ('socks-protocol', SocksEditor, 35), + ('xray-tun-settings', XrayTunSettingsDialog, 25), + ('hysteria2-tun-settings', Hysteria2TunSettingsDialog, 25), + ( + 'routing-rule', + lambda: RoutingRuleEditDialog( + {'type': 'field', 'outboundTag': 'proxy'} + ), + 30, + ), + ( + 'routing-rules', + lambda: RoutingRulesDialog({'rules': []}), + 50, + ), + ( + 'routing-preview', + lambda: RoutingPreviewDialog({'rules': []}), + 50, + ), + ('subscription-editor', _SubscriptionEditorDialog, 50), + ) + + with isolatedSettings(): + for name, factory, iterations in factories: + with self.subTest(family=name): + references, destroyed = [], [] + + 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() + + self.assertAllDestroyed( + references, + destroyed, + iterations, + ) + + def testQRCodeTopLevelWindowIsDeletedOnClose(self): + """Exercise the dedicated transient AppQMainWindow policy 100 times.""" + iterations = 100 + references, destroyed = [], [] + + for _index in range(iterations): + window = QRCodeWindow() + window.destroyed.connect(lambda *_args: destroyed.append(True)) + + references.append(weakref.ref(window)) + + window.show() + window.close() + + del window + + collectAtBoundary() + + self.assertAllDestroyed(references, destroyed, iterations) + + def testTextEditorWindowIsIntentionallyReusableThenExplicitlyDestroyed(self): + """Reuse one persistent editor without duplicating menus or actions.""" + with isolatedSettings(): + editor = TextEditorWindow() + reference = weakref.ref(editor) + actionCount = len(editor.actions()) + fileActionCount = len(editor.fileMenu.actions()) + + for _index in range(50): + editor.show() + + processQtEvents(1) + + editor.close() + + processQtEvents(1) + + self.assertTrue(isValid(editor)) + self.assertEqual(len(editor.actions()), actionCount) + self.assertEqual(len(editor.fileMenu.actions()), fileActionCount) + + self.assertNotIn(editor._lifetimeKey, AppQMainWindow._openWindows) + + editor.deleteLater() + + del editor + + collectAtBoundary() + + self.assertTrue(waitFor(lambda: reference() is None)) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_qt_stress.py b/tests/test_qt_stress.py new file mode 100644 index 0000000..1365889 --- /dev/null +++ b/tests/test_qt_stress.py @@ -0,0 +1,214 @@ +# 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 . + +"""Measure lifetime and memory trends across hundreds of Qt UI cycles.""" + +from __future__ import annotations + +from Furious.Frozenlib import Mixins +from Furious.Qt import AppQAction, AppQMenu, AppQTransientDialog + +from PySide6 import QtCore + +from tests.support import ( + application, + collectAtBoundary, + currentRSS, + qObjectCount, +) + +import unittest +import weakref +import tracemalloc + + +class StressDialog(AppQTransientDialog): + """Own representative actions, a menu, a timer, and signal callbacks.""" + + def __init__(self): + """Create one inexpensive but ownership-rich transient dialog.""" + super().__init__() + + self.timer = QtCore.QTimer(self) + self.timer.timeout.connect(self.update) + self.timer.start(1000) + + self.actionsForTest = tuple( + AppQAction(f'Action {index}', callback=self.update) for index in range(3) + ) + + self.menuForTest = AppQMenu(*self.actionsForTest, parent=self) + + +class QtMemoryStressTest(unittest.TestCase): + """Reject linear live-object retention while tolerating allocator caching.""" + + WarmupIterations = 40 + BatchIterations = 100 + BatchCount = 3 + + @classmethod + def setUpClass(cls): + """Create the process-wide headless QApplication.""" + application() + + @staticmethod + def _runBatch(iterations): + """Create and close a batch without forcing collection per cycle.""" + references, destroyed = [], [] + + for _index in range(iterations): + dialog = StressDialog() + dialog.destroyed.connect(lambda *_args: destroyed.append(True)) + + references.append(weakref.ref(dialog)) + + dialog.show() + dialog.close() + + del dialog + + collectAtBoundary() + + return references, len(destroyed) + + def testThreeHundredCyclesPlateauAfterWarmup(self): + """Combine direct destruction, pool counts, Python memory, and RSS trend.""" + collectAtBoundary() + + baseline = { + 'dialog': qObjectCount(StressDialog), + 'timer': qObjectCount(QtCore.QTimer), + 'menu': qObjectCount(AppQMenu), + 'action': qObjectCount(AppQAction), + 'translationPool': len(Mixins.QTranslatable.ObjectsPool), + 'themePool': len(Mixins.ThemeAware.ObjectsPool), + 'connectionPool': len(Mixins.ConnectionAware.ObjectsPool), + } + + tracemalloc.start() + + try: + warmupReferences, warmupDestroyed = self._runBatch(self.WarmupIterations) + warmupPython = tracemalloc.get_traced_memory()[0] + rssSamples = [currentRSS()] + pythonSamples = [warmupPython] + liveSamples = [qObjectCount(StressDialog)] + resourceSamples = [ + { + 'timer': qObjectCount(QtCore.QTimer), + 'menu': qObjectCount(AppQMenu), + 'action': qObjectCount(AppQAction), + } + ] + references = list(warmupReferences) + destroyed = warmupDestroyed + + for _batch in range(self.BatchCount): + batchReferences, batchDestroyed = self._runBatch(self.BatchIterations) + references.extend(batchReferences) + destroyed += batchDestroyed + rssSamples.append(currentRSS()) + pythonSamples.append(tracemalloc.get_traced_memory()[0]) + liveSamples.append(qObjectCount(StressDialog)) + resourceSamples.append( + { + 'timer': qObjectCount(QtCore.QTimer), + 'menu': qObjectCount(AppQMenu), + 'action': qObjectCount(AppQAction), + } + ) + + totalIterations = ( + self.WarmupIterations + self.BatchIterations * self.BatchCount + ) + + self.assertEqual(destroyed, totalIterations) + self.assertTrue(all(reference() is None for reference in references)) + self.assertEqual(liveSamples, [baseline['dialog']] * 4) + + for resourceSample in resourceSamples: + self.assertEqual(resourceSample['timer'], baseline['timer']) + self.assertEqual(resourceSample['menu'], baseline['menu']) + self.assertEqual(resourceSample['action'], baseline['action']) + + self.assertEqual( + len(Mixins.QTranslatable.ObjectsPool), + baseline['translationPool'], + ) + self.assertEqual( + len(Mixins.ThemeAware.ObjectsPool), + baseline['themePool'], + ) + self.assertEqual( + len(Mixins.ConnectionAware.ObjectsPool), + baseline['connectionPool'], + ) + + pythonGrowth = tuple( + later - earlier + for earlier, later in zip(pythonSamples, pythonSamples[1:]) + ) + # A persistent leak produces similar positive growth in every + # post-warm-up batch. Permit normal tracing/allocator noise, but + # fail a sustained multi-megabyte trend correlated with no cache + # stabilization. + suspiciousPythonGrowth = ( + all(growth > 512 * 1024 for growth in pythonGrowth) + and sum(pythonGrowth) > 3 * 1024 * 1024 + ) + + self.assertFalse( + suspiciousPythonGrowth, + f'continued Python retention after warm-up: {pythonSamples}', + ) + + presentRSS = tuple(value for value in rssSamples if value is not None) + + if len(presentRSS) == len(rssSamples): + rssGrowth = tuple( + later - earlier + for earlier, later in zip(presentRSS, presentRSS[1:]) + ) + suspiciousRSSGrowth = ( + all(growth > 4 * 1024 * 1024 for growth in rssGrowth) + and sum(rssGrowth) > 16 * 1024 * 1024 + ) + + self.assertFalse( + suspiciousRSSGrowth, + f'continued resident-memory growth after warm-up: {presentRSS}', + ) + + print( + 'Qt lifetime stress:', + { + 'widget': StressDialog.__name__, + 'iterations': totalIterations, + 'destroyed': destroyed, + 'liveSamples': liveSamples, + 'resourceSamples': resourceSamples, + 'pythonBytes': pythonSamples, + 'rssBytes': rssSamples, + }, + ) + finally: + tracemalloc.stop() + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_socks_uri.py b/tests/test_socks_uri.py new file mode 100644 index 0000000..7fad2cb --- /dev/null +++ b/tests/test_socks_uri.py @@ -0,0 +1,306 @@ +# 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 interoperable and centralized SOCKS share-link handling.""" + +from __future__ import annotations + +from Furious.Backends.Configuration import ConfigXray +from Furious.Backends.SocksURI import ( + SocksURIData, + SocksURIError, + parseSocksURI, + serializeSocksURI, +) +from Furious.Backends.Xray.Protocols import XRAY_PROTOCOL_HANDLERS +from Furious.Extensions.StandardSubscriptions import StandardSubscriptionPlugin +from Furious.Plugins.API import FuriousPlugin, PluginMetadata +from Furious.Plugins.Registry import PluginRegistry +from Furious.Service.SubscriptionImporter import ( + SubscriptionImportService, + SubscriptionSource, +) + +from urllib.parse import quote + +import base64 +import unittest + + +class SocksURICodecTest(unittest.TestCase): + """Exercise canonical export and compatibility import forms.""" + + def testBasicHostnameAndIPv4Imports(self): + """Parse unauthenticated hostnames and IPv4 endpoints.""" + self.assertEqual( + parseSocksURI('socks://example.com:1080'), + SocksURIData('example.com', 1080), + ) + self.assertEqual( + parseSocksURI('socks5://127.0.0.1:1081#Local'), + SocksURIData('127.0.0.1', 1081, tag='Local'), + ) + + def testDirectCredentialsUseStrictPercentDecoding(self): + """Preserve reserved characters without form-style plus decoding.""" + uri = ( + 'socks://user%40name:' + 'p%3Aa%23ss%25word%2Fwith%20space%2Bplus@example.com:1080' + '#My%20SOCKS%20%23%25%20服务器' + ) + + self.assertEqual( + parseSocksURI(uri), + SocksURIData( + 'example.com', + 1080, + 'user@name', + 'p:a#ss%word/with space+plus', + 'My SOCKS #% 服务器', + ), + ) + + def testV2rayNBase64UserinfoCompatibility(self): + """Accept v2rayN's standard and URL-safe Base64 userinfo.""" + value = SocksURIData('example.com', 1080, 'user', 'pass', 'Remark') + + self.assertEqual( + parseSocksURI('socks://dXNlcjpwYXNz@example.com:1080#Remark'), + value, + ) + + credentials = base64.urlsafe_b64encode('用户:密码/@'.encode('utf-8')).decode( + 'ascii' + ) + uri = f'socks5://{quote(credentials, safe="")}@example.com:1080' + + self.assertEqual( + parseSocksURI(uri), + SocksURIData('example.com', 1080, '用户', '密码/@'), + ) + + def testLegacyWholePayloadCompatibility(self): + """Accept the historical Base64 credentials-and-endpoint payload.""" + payload = base64.b64encode(b'user:pass@example.com:1080').decode('ascii') + + self.assertEqual( + parseSocksURI(f'socks://{payload}#Legacy%20Node'), + SocksURIData('example.com', 1080, 'user', 'pass', 'Legacy Node'), + ) + + ipv6Payload = base64.b64encode(b'user:pass@2001:db8::1:1080').decode('ascii') + + self.assertEqual( + parseSocksURI(f'socks://{ipv6Payload}'), + SocksURIData('2001:db8::1', 1080, 'user', 'pass'), + ) + + def testIPv6RoundTripUsesBrackets(self): + """Keep IPv6 colons separate from the URI port delimiter.""" + value = SocksURIData('2001:0db8::1', 1080, 'user', 'pass', 'IPv6') + uri = serializeSocksURI(value) + + self.assertEqual( + uri, + 'socks://user:pass@[2001:db8::1]:1080#IPv6', + ) + self.assertEqual( + parseSocksURI(uri), + SocksURIData('2001:db8::1', 1080, 'user', 'pass', 'IPv6'), + ) + + def testCanonicalExportUsesInteroperableSocksAuthority(self): + """Export one socks scheme with unambiguous percent-encoded userinfo.""" + self.assertEqual( + serializeSocksURI( + SocksURIData('example.com', 1080, 'user', 'pass', 'My SOCKS') + ), + 'socks://user:pass@example.com:1080#My%20SOCKS', + ) + self.assertEqual( + serializeSocksURI(SocksURIData('example.com', 1080)), + 'socks://example.com:1080', + ) + + def testUnicodeCredentialsAndTagRoundTrip(self): + """Round-trip Unicode through UTF-8 Base64 and percent encoding.""" + value = SocksURIData( + 'xn--fsqu00a.xn--0zwm56d', + 1080, + '用户', + '密码:@#% /', + '测试 SOCKS #%', + ) + + self.assertEqual(parseSocksURI(serializeSocksURI(value)), value) + + def testCompatibilitySchemeAliasesAreCaseInsensitive(self): + """Accept existing Furious aliases and v2rayN's SOCKS4 alias.""" + for scheme in ('SOCKS', 'socks5', 'socks5h', 'socks4'): + with self.subTest(scheme=scheme): + self.assertEqual( + parseSocksURI(f'{scheme}://example.com:1080'), + SocksURIData('example.com', 1080), + ) + + def testMalformedLinksFailCleanly(self): + """Reject corrupt authority, credentials, and URI components.""" + invalid = ( + '', + 'http://example.com:1080', + 'socks://:1080', + 'socks://example.com', + 'socks://example.com:0', + 'socks://example.com:65536', + 'socks://example.com:not-a-port', + 'socks://[2001:db8::1:1080', + 'socks://not-base64@example.com:1080', + 'socks://user%ZZ:pass@example.com:1080', + 'socks://example.com:1080/path', + 'socks://example.com:1080?unsupported=1', + 'socks://example.com:1080#bad%', + ) + + for uri in invalid: + with self.subTest(uri=uri): + with self.assertRaises(SocksURIError): + parseSocksURI(uri) + + +class SocksURIIntegrationTest(unittest.TestCase): + """Verify the codec at Xray configuration and plugin boundaries.""" + + def testConfigImportPreservesRuntimeFieldsAndRemark(self): + """Pass imported credentials into Xray's SOCKS outbound settings.""" + handler = next( + item for item in XRAY_PROTOCOL_HANDLERS if item.descriptor.id == 'SOCKS' + ) + result = handler.parse( + 'socks://user%40name:p%3Aass@[2001:db8::1]:1080#My%20Node' + ) + config = result.configuration + + self.assertEqual(config.proxyProtocol, 'socks') + self.assertEqual( + config.proxyServerObject, + { + 'address': '2001:db8::1', + 'port': 1080, + 'user': 'user@name', + 'pass': 'p:ass', + }, + ) + self.assertEqual(result.metadata['displayName'], 'My Node') + + def testConfigExportAndReimportPreserveSemantics(self): + """Share a persisted SOCKS outbound through the centralized codec.""" + original = ConfigXray( + { + 'outbounds': [ + { + 'tag': 'proxy', + 'protocol': 'socks', + 'settings': { + 'address': 'example.com', + 'port': 1080, + 'user': 'user:@', + 'pass': 'password #%', + }, + } + ] + } + ) + uri = original.toURI('共享 SOCKS') + handler = next( + item for item in XRAY_PROTOCOL_HANDLERS if item.descriptor.id == 'SOCKS' + ) + result = handler.parse(uri) + reparsed = result.configuration + + self.assertTrue(uri.startswith('socks://')) + self.assertEqual(reparsed.proxyServerObject, original.proxyServerObject) + self.assertEqual(result.metadata['displayName'], '共享 SOCKS') + + def testProtocolCapabilityOwnsAllAliasesAndValidatesEndpoint(self): + """Keep URI dispatch and semantic validation inside the SOCKS handler.""" + handler = next( + item for item in XRAY_PROTOCOL_HANDLERS if item.descriptor.id == 'SOCKS' + ) + + self.assertEqual( + set(handler.schemes), + {'socks', 'socks5', 'socks5h', 'socks4'}, + ) + + parsed = handler.parse('socks5://dXNlcjpwYXNz@example.com:1080#Node') + + self.assertIsNotNone(parsed) + self.assertEqual(parsed.metadata['displayName'], 'Node') + self.assertEqual(handler.validate(parsed.configuration), tuple()) + + parsed.configuration.proxyServerObject['port'] = 0 + + self.assertIn( + 'server port must be between 1 and 65535', + handler.validate(parsed.configuration), + ) + + def testLineSubscriptionUsesTheRegisteredSocksCodec(self): + """Route subscription links through the same protocol capability.""" + handler = next( + item for item in XRAY_PROTOCOL_HANDLERS if item.descriptor.id == 'SOCKS' + ) + + class SocksOnlyPlugin(FuriousPlugin): + """Register only the SOCKS capability needed by this fixture.""" + + metadata = PluginMetadata('test.socks', 'Test SOCKS') + capabilities = (handler,) + + registry = PluginRegistry() + registry.register(SocksOnlyPlugin()) + registry.register(StandardSubscriptionPlugin()) + + try: + result = SubscriptionImportService(registry).importPayload( + b'socks://dXNlcjpwYXNz@example.com:1080#Subscribed%20SOCKS', + SubscriptionSource('fixture'), + ) + + self.assertIsNotNone(result) + self.assertEqual(result.rejectedItems, 0) + self.assertEqual(len(result.profiles), 1) + self.assertEqual(result.profiles[0].itemRemark, 'Subscribed SOCKS') + self.assertEqual( + result.profiles[0].connection.proxyServerObject, + { + 'address': 'example.com', + 'port': 1080, + 'user': 'user', + 'pass': 'pass', + }, + ) + self.assertEqual( + registry.exportConfig(result.profiles[0]), + 'socks://user:pass@example.com:1080#Subscribed%20SOCKS', + ) + finally: + registry.shutdown() + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_ui_behavior.py b/tests/test_ui_behavior.py new file mode 100644 index 0000000..24f5e86 --- /dev/null +++ b/tests/test_ui_behavior.py @@ -0,0 +1,265 @@ +# 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 high-value Qt presentation and editor integration boundaries.""" + +from __future__ import annotations + +from Furious.Backends.ExternalCore.Configuration import ( + BLANK_CONFIG_EXTERNAL_CORE, + ConfigExternalCore, +) +from Furious.Backends.ExternalCore.Editor import ExternalCoreEditor +from Furious.Backends.Xray.RoutingWindow import RoutingRulesDialog +from Furious.Models import ProfileMetadata, ServerProfile +from Furious.Qt import AppQMessageBox +from Furious.Service import ( + APPLICATION_LOG_CATEGORY, + CORE_LOG_CATEGORY, + LogManager, +) +from Furious.Window.LogPage import LogPage +from Furious.Window.SubscriptionPage import _SubscriptionEditorDialog + +from tests.support import ( + application, + collectAtBoundary, + isolatedSettings, + processQtEvents, +) + +import copy +import unittest + + +class EditorMappingTest(unittest.TestCase): + """Verify editor fields preserve structured configuration semantics.""" + + @classmethod + def setUpClass(cls): + """Create the process-wide headless QApplication.""" + application() + + def tearDown(self): + """Finish every deferred transient deletion between tests.""" + collectAtBoundary() + + def testExternalCoreEditorRoundTripsStructuredFields(self): + """Keep arguments, environment, process paths, and TUN data distinct.""" + configuration = ConfigExternalCore(copy.deepcopy(BLANK_CONFIG_EXTERNAL_CORE)) + configuration.update( + { + 'executable': 'C:/Program Files/Fixture/core.exe', + 'workingDirectory': 'C:/Program Files/Fixture', + 'arguments': ['--config', 'name with spaces.json'], + 'environment': {'TOKEN': 'one=two', 'UNICODE': '测试'}, + 'useApplicationTun2socks': True, + 'tunRemoteAddress': '2001:db8::42', + } + ) + profile = ServerProfile.fromConfiguration( + configuration, + ProfileMetadata(displayName='Fixture core'), + ) + + with isolatedSettings(): + editor = ExternalCoreEditor() + editor.factoryToInput(profile) + + self.assertEqual( + editor._argumentsInput.values(), + ['--config', 'name with spaces.json'], + ) + self.assertEqual( + editor._environmentInput.values(), + {'TOKEN': 'one=two', 'UNICODE': '测试'}, + ) + self.assertTrue(editor._applicationTun2socksInput.isChecked()) + self.assertTrue(editor._tunRemoteAddressInput.widgets()[1].isEnabled()) + self.assertEqual(editor._tunRemoteAddressInput.text(), '2001:db8::42') + self.assertEqual(len(editor.groupBoxSequence()), 1) + self.assertAlmostEqual(editor.height() / editor.width(), 1.618, places=2) + + editor._argumentsInput._input.setText( + '--mode direct --label "a value with spaces"' + ) + editor._environmentInput._input.setPlainText('A=1\nB=two=three') + editor._tunRemoteAddressInput._input.setText('server.example.com') + + self.assertTrue(editor.inputToFactory(profile)) + self.assertEqual( + profile.connection['arguments'], + ['--mode', 'direct', '--label', 'a value with spaces'], + ) + self.assertEqual( + profile.connection['environment'], + {'A': '1', 'B': 'two=three'}, + ) + self.assertEqual( + profile.connection.tunRemoteAddress(), + 'server.example.com', + ) + + editor.close() + + def testSubscriptionEditorNormalizesPresentationValues(self): + """Return one complete subscription record from its visual controls.""" + with isolatedSettings(): + dialog = _SubscriptionEditorDialog( + { + 'remark': ' Fixture subscription ', + 'webURL': 'https://example.test/subscription', + 'enabled': False, + 'autoupdate': 'Every 6 hours', + 'proxy': 'Direct', + 'userAgent': ' Fixture/1.0 ', + 'filter': ' keep.* ', + } + ) + + values = dialog.subscription() + + self.assertEqual(values['remark'], 'Fixture subscription') + self.assertEqual(values['webURL'], 'https://example.test/subscription') + self.assertFalse(values['enabled']) + self.assertEqual(values['userAgent'], 'Fixture/1.0') + self.assertEqual(values['filter'], 'keep.*') + + dialog.accept() + + self.assertEqual( + dialog.result(), + _SubscriptionEditorDialog.DialogCode.Accepted, + ) + + +class UnifiedLogPageTest(unittest.TestCase): + """Prove bounded collection is eager while hidden-page rendering is lazy.""" + + @classmethod + def setUpClass(cls): + """Create the process-wide headless QApplication.""" + application() + + def testHiddenPageRendersOneOrderedSnapshotWhenShown(self): + """Do not mutate the document while hidden; catch up exactly once.""" + with isolatedSettings(): + manager = LogManager(maximumEntries=5) + page = LogPage(manager=manager) + + manager.append('application one', APPLICATION_LOG_CATEGORY) + manager.append('core one', CORE_LOG_CATEGORY) + + processQtEvents() + + self.assertEqual(page.textBrowser.toPlainText(), '') + self.assertTrue(page._entriesDirty) + + page.show() + + processQtEvents() + + self.assertEqual( + page.textBrowser.toPlainText().splitlines(), + ['application one', 'core one'], + ) + + page.hide() + manager.append('core two', CORE_LOG_CATEGORY) + + processQtEvents() + + self.assertNotIn('core two', page.textBrowser.toPlainText()) + + page.show() + + processQtEvents() + + self.assertEqual( + page.textBrowser.toPlainText().splitlines(), + ['application one', 'core one', 'core two'], + ) + + coreIndex = page.filterComboBox.findData(CORE_LOG_CATEGORY) + page.filterComboBox.setCurrentIndex(coreIndex) + + processQtEvents() + + self.assertEqual( + page.textBrowser.toPlainText().splitlines(), + ['core one', 'core two'], + ) + + page.close() + page.deleteLater() + + +class DialogBehaviorTest(unittest.TestCase): + """Exercise no-selection guards and QMessageBox-compatible results.""" + + @classmethod + def setUpClass(cls): + """Create the process-wide headless QApplication.""" + application() + + def tearDown(self): + """Finish every deferred transient deletion between tests.""" + collectAtBoundary() + + def testRoutingRulesActionsStayEnabledAndNoSelectionIsSafe(self): + """Keep the compact top actions visible without creating a warning.""" + dialog = RoutingRulesDialog({'rules': []}) + + self.assertTrue(dialog.addButton.isEnabled()) + self.assertTrue(dialog.deleteButton.isEnabled()) + self.assertTrue(dialog.closeWindowButton.isEnabled()) + self.assertIsNotNone(dialog.layout().itemAt(0).layout()) + self.assertIs(dialog.layout().itemAt(1).widget(), dialog.listWidget) + + dialog.deleteRule() + dialog.editRule() + + self.assertEqual(AppQMessageBox._openMessageBoxes, {}) + self.assertEqual(dialog.routing['rules'], []) + + dialog.closeWindowButton.click() + + def testMessageBoxButtonPublishesCompatibleResult(self): + """Emit one clicked button and finish with its standard-button value.""" + messageBox = AppQMessageBox( + text='Continue?', + buttons=( + AppQMessageBox.StandardButton.Yes | AppQMessageBox.StandardButton.No + ), + ) + clicked = [] + finished = [] + messageBox.buttonClicked.connect(clicked.append) + messageBox.finished.connect(finished.append) + yesButton = messageBox.button(AppQMessageBox.StandardButton.Yes) + + yesButton.click() + + processQtEvents() + + self.assertEqual(clicked, [yesButton]) + self.assertEqual(finished, [int(AppQMessageBox.StandardButton.Yes)]) + self.assertIs(messageBox.clickedButton(), yesButton) + + +if __name__ == '__main__': + unittest.main()