diff --git a/Furious/Backends/Hysteria1/Plugin.py b/Furious/Backends/Hysteria1/Plugin.py index 17a368d..99b3749 100644 --- a/Furious/Backends/Hysteria1/Plugin.py +++ b/Furious/Backends/Hysteria1/Plugin.py @@ -94,6 +94,7 @@ class Hysteria1CoreRuntimeFactory(CoreRuntimeFactory): Hysteria1.mmdb(routingObject.get('mmdb', '')), ), options=request.options, + startup=CoreRuntimeStartup(endpoint=config.httpProxy()), ) def prepareDownloadTest(self, config, port: int): diff --git a/Furious/Backends/Hysteria2/Plugin.py b/Furious/Backends/Hysteria2/Plugin.py index 2ecf13f..c4b034f 100644 --- a/Furious/Backends/Hysteria2/Plugin.py +++ b/Furious/Backends/Hysteria2/Plugin.py @@ -211,6 +211,7 @@ class Hysteria2CoreRuntimeFactory(CoreRuntimeFactory): runtime, request.configuration, options=request.options, + startup=CoreRuntimeStartup(endpoint=request.configuration.httpProxy()), ) def prepareDownloadTest(self, config, port: int): diff --git a/Furious/Backends/Xray/Plugin.py b/Furious/Backends/Xray/Plugin.py index b9add89..738d06a 100644 --- a/Furious/Backends/Xray/Plugin.py +++ b/Furious/Backends/Xray/Plugin.py @@ -327,7 +327,12 @@ class XrayCoreRuntimeFactory(CoreRuntimeFactory): ) runtime.xrayStatsTarget = statsTarget - return CoreRuntimeLaunch(runtime, config, options=request.options) + return CoreRuntimeLaunch( + runtime, + config, + options=request.options, + startup=CoreRuntimeStartup(endpoint=config.httpProxy()), + ) def prepareDownloadTest(self, config, port: int): """Create an Xray configuration with one local HTTP test inbound.""" diff --git a/Furious/Controllers/ConnectionController.py b/Furious/Controllers/ConnectionController.py index 3424f96..bcd44d7 100644 --- a/Furious/Controllers/ConnectionController.py +++ b/Furious/Controllers/ConnectionController.py @@ -24,6 +24,7 @@ from Furious.Interface import * from Furious.Models import ServerProfile from Furious.Plugins import getPluginRegistry from Furious.Qt.DynamicTranslate import gettext as _ +from Furious.Qt.Signals import connectWeakly from Furious.Repository import Storage from Furious.Service import ( CORE_LOG_CATEGORY, @@ -102,6 +103,8 @@ class ConnectionController(QtCore.QObject): self._state = ConnectionState.Disconnected self._activeProfile = None self._lastError = None + self._startOperation = None + self._pendingHttpProxy = '' self._actionTimer = QtCore.QTimer(self) self._actionTimer.timeout.connect(self._callActionFromQueue) @@ -187,6 +190,8 @@ class ConnectionController(QtCore.QObject): def _reset(self): """Restore disconnected state after all runtime resources stop.""" + self._startOperation = None + self._pendingHttpProxy = '' self.progressFinished.emit(True) self._setActiveProfile(None) @@ -269,12 +274,57 @@ class ConnectionController(QtCore.QObject): self._lastError = None self._setActiveProfile(configuration) + self._pendingHttpProxy = httpProxy self._startConnecting() logManager = AppLogManager() # Retain application diagnostics while starting a fresh runtime log. logManager.clear(runtimeOnly=True) + startAsync = getattr(self._coreManager, 'startAsync', None) + + if callable(startAsync): + try: + operation = startAsync( + configuration, + routing=AppSettings.get('Routing'), + exitCallback=self.coreExitCallback, + msgCallbackCore=logManager.callback(CORE_LOG_CATEGORY), + msgCallbackTUN_=logManager.callback( + TUN2SOCKS_LOG_CATEGORY, + source='Tun2socks', + ), + ) + except Exception as ex: + logger.error(f'failed to schedule core manager startup: {ex}') + + return self._failConnection( + f'{configuration.coreName()}: ' + _('Unknown error'), + str(ex), + ) + + self._startOperation = operation + connectWeakly( + operation.succeeded, + self, + '_connectionStartSucceeded', + sender=operation, + ) + connectWeakly( + operation.failed, + self, + '_connectionStartFailed', + sender=operation, + ) + connectWeakly( + operation.cancelled, + self, + '_connectionStartCancelled', + sender=operation, + ) + + return True + startExceptionDetails = '' try: @@ -315,6 +365,10 @@ class ConnectionController(QtCore.QObject): startExceptionDetails, ) + return self._finishConnection(configuration, httpProxy) + + def _finishConnection(self, configuration, httpProxy) -> bool: + """Commit system integration after manager runtime ownership commits.""" settings = AppSettings.get('CustomProxyBypass') proxyServerBypass = ( @@ -352,14 +406,80 @@ class ConnectionController(QtCore.QObject): return True + @QtCore.Slot(object) + def _connectionStartSucceeded(self, operation): + """Finish the exact manager generation that committed successfully.""" + if operation is not self._startOperation or not self.isConnecting(): + return + + self._startOperation = None + self._emitRuntimesChanged() + + while not self._actionQueue.empty(): + self._callActionFromQueue() + + if not self.isConnecting(): + return + + configuration = self.activeProfile + + if configuration is None: + self._failConnection(_('Unknown error')) + + return + + self._finishConnection(configuration, self._pendingHttpProxy) + + @QtCore.Slot(object, str, str) + def _connectionStartFailed(self, operation, message, details): + """Return one failed manager generation to stable disconnected state.""" + if operation is not self._startOperation: + return + + self._startOperation = None + self._emitRuntimesChanged() + + configuration = self.activeProfile + coreName = configuration.coreName() if configuration is not None else '' + startError = message or getattr(self._coreManager, 'lastStartError', '') + displayMessage = (f'{coreName}: ' if coreName else '') + ( + _(startError) if startError else _('Unknown error') + ) + + self._failConnection(displayMessage, details) + + @QtCore.Slot(object) + def _connectionStartCancelled(self, operation): + """Ignore stale cancellation or reset an externally cancelled start.""" + if operation is not self._startOperation: + return + + self._startOperation = None + self._emitRuntimesChanged() + self._reset() + def startDisconnection(self, notification: str = '') -> bool: """Stop the active runtime and optionally request a notification.""" if self.state is ConnectionState.Disconnected: return False + operation = self._startOperation + + self._startOperation = None self._setState(ConnectionState.Disconnecting) self._actionTimer.stop() + if operation is not None: + cancelStart = getattr(self._coreManager, 'cancelStart', None) + + if callable(cancelStart): + try: + cancelStart(operation) + except Exception as ex: + # Any non-exit exceptions + + logger.error(f'failed to cancel connection startup: {ex}') + try: SystemProxy.off() except Exception as ex: @@ -390,13 +510,10 @@ class ConnectionController(QtCore.QObject): def startReconnection(self, notification: str = '') -> bool: """Restart the active repository profile when lifecycle state permits.""" - if self.state in ( - ConnectionState.Connecting, - ConnectionState.Disconnecting, - ): + if self.state is ConnectionState.Disconnecting: return False - if self.isConnected(): + if self.isConnected() or self.isConnecting(): self.startDisconnection(notification) return self.startConnection() diff --git a/Furious/Core/CoreProcessWorker.py b/Furious/Core/CoreProcessWorker.py index 4460d00..58dd114 100644 --- a/Furious/Core/CoreProcessWorker.py +++ b/Furious/Core/CoreProcessWorker.py @@ -64,7 +64,12 @@ class CoreProcessState(Enum): @dataclass class CoreLaunchSpec: - """Describe the parameters required by a core launch operation.""" + """Describe the parameters required by a core launch operation. + + 'waitCore' preserves the synchronous plugin compatibility path. Normal GUI + connections launch with it disabled and confirm readiness through the + connection manager's event-driven startup transaction. + """ DefaultWaitTime: ClassVar[int] = 2500 @@ -437,20 +442,35 @@ class CoreProcessWorker(CoreProcessMonitor, ABC): self.msgQueue.setTimeout(MsgQueue.ACTIVE_DRAIN_INTERVAL) self.msgQueue.startTimer() + # Monitor the process during asynchronous startup as well. A child that + # exits before readiness must fail the owning transaction instead of + # waiting for a later connected-state health check. + self.daemon.start(CORE_CHECK_ALIVE_INTERVAL) + if launchSpec.waitCore: # Wait for the core to start up completely PySide6Legacy.eventLoopWait(launchSpec.waitTime) if self.queryIsAlive(): + # Preserve the low-level compatibility contract: a successfully + # spawned process is Running even when its caller disables the + # historical grace wait. The connection transaction separately + # owns semantic readiness and final commit. self.setState(CoreProcessState.Running) - # Start core daemon - self.daemon.start(CORE_CHECK_ALIVE_INTERVAL) - return True else: return False + def confirmStartup(self) -> bool: + """Move a live externally observed process into the running state.""" + if not self.isAlive(): + return False + + self.setState(CoreProcessState.Running) + + return True + def stop(self): """Stop the core process worker.""" self.msgQueue.stopTimer() diff --git a/Furious/Interface/Runtime.py b/Furious/Interface/Runtime.py index c0bbe3c..77cbe45 100644 --- a/Furious/Interface/Runtime.py +++ b/Furious/Interface/Runtime.py @@ -69,6 +69,14 @@ class CoreRuntime(ABC): if callable(self._exitCallback): self._exitCallback(self, exitcode) + def setExitCallback(self, callback): + """Transfer termination reporting to the runtime's current owner.""" + self._exitCallback = callback + + def confirmStartup(self) -> bool: + """Confirm readiness after an external startup observer succeeds.""" + return True + def startError(self) -> str: """Return the most recent concise startup failure, if any.""" return self._startError diff --git a/Furious/Plugins/API.py b/Furious/Plugins/API.py index fc3491c..c1064dd 100644 --- a/Furious/Plugins/API.py +++ b/Furious/Plugins/API.py @@ -30,6 +30,7 @@ __all__ = [ 'CoreRuntimeFactory', 'CoreRuntimeLaunch', 'CoreRuntimeRequest', + 'CoreRuntimeStartup', 'FuriousPlugin', 'NavigationPageDescriptor', 'NavigationPageProvider', @@ -401,6 +402,15 @@ class CoreRuntimeRequest: options: Mapping[str, Any] = field(default_factory=dict) +@dataclass(frozen=True) +class CoreRuntimeStartup: + """Describe an optional event-driven readiness contract for one runtime.""" + + endpoint: str = '' + timeout: int = 2500 + retryInterval: int = 50 + + @dataclass(frozen=True) class CoreRuntimeLaunch: """Bind a constructed core runtime to its prepared start arguments.""" @@ -409,14 +419,18 @@ class CoreRuntimeLaunch: configuration: Any arguments: Tuple[Any, ...] = tuple() options: Mapping[str, Any] = field(default_factory=dict) + startup: Optional[CoreRuntimeStartup] = None - def start(self) -> bool: + def start(self, **optionOverrides) -> bool: """Start the prepared core runtime.""" + options = dict(self.options) + options.update(optionOverrides) + return bool( self.runtime.start( self.configuration, *self.arguments, - **dict(self.options), + **options, ) ) diff --git a/Furious/Plugins/__init__.py b/Furious/Plugins/__init__.py index d468c6d..1c163a0 100644 --- a/Furious/Plugins/__init__.py +++ b/Furious/Plugins/__init__.py @@ -26,6 +26,7 @@ from .API import ( CoreRuntimeFactory, CoreRuntimeLaunch, CoreRuntimeRequest, + CoreRuntimeStartup, FuriousPlugin, NavigationPageDescriptor, NavigationPageProvider, @@ -74,6 +75,7 @@ __all__ = [ 'CoreRuntimeFactory', 'CoreRuntimeLaunch', 'CoreRuntimeRequest', + 'CoreRuntimeStartup', 'FuriousPlugin', 'NavigationPageDescriptor', 'NavigationPageProvider', diff --git a/Furious/Service/AGENTS.md b/Furious/Service/AGENTS.md index bd0ea7b..a6f5fdc 100644 --- a/Furious/Service/AGENTS.md +++ b/Furious/Service/AGENTS.md @@ -12,8 +12,11 @@ ## Runtime and asynchronous invariants - `ConnectionManager` consumes an attempt-scoped copy, asks the selected factory for native-TUN/application-tun2socks - policy, and owns exact runtimes only after commit. On failure it rolls back only resources acquired by that attempt and - restores host routing/DNS state through those owners. + policy, and owns exact runtimes only after commit. Normal built-in GUI startup belongs to one generation-checked + `ConnectionStartOperation`: launch, local-endpoint readiness, asynchronous DNS, TUN-device observation, host mutation, + rollback, and commit advance through owned Qt timers/signals without a nested event loop. The synchronous `start()` + path is a compatibility boundary for legacy plugins/non-interactive callers, not the preferred GUI path. On failure + or cancellation, roll back only resources acquired by that attempt and restore host routing/DNS through those owners. - Every async workflow defines supersession: generation/version checks for stale completion or exact-object identity for independent requests. Terminal paths release context, finish/abort once, and schedule each Qt reply for deletion. Callbacks cannot retain a shut-down manager; worker results cross into the manager's Qt thread before mutation. diff --git a/Furious/Service/ConnectionManager.py b/Furious/Service/ConnectionManager.py index 173b2b4..cd9df7e 100644 --- a/Furious/Service/ConnectionManager.py +++ b/Furious/Service/ConnectionManager.py @@ -31,24 +31,33 @@ from Furious.Frozenlib import ( SystemRoutingTable, SystemRuntime, isValidIPAddress, + parseHostPort, ) from Furious.Interface import CoreRuntime from Furious.Models import CoreConfiguration, ServerProfile from Furious.Repository import Storage -from Furious.Plugins import TUNPreparationError, getPluginRegistry +from Furious.Plugins import ( + CoreRuntimeLaunch, + CoreRuntimeStartup, + TUNPreparationError, + getPluginRegistry, +) +from Furious.Qt.Signals import connectWeakly, singleShotWeakly from Furious.Service.DnsResolver import DnsResolver from Furious.Core.CoreProcessWorker import CoreProcessWorker from Furious.Core.Tun2socks import Tun2socks from typing import Callable, Tuple, Union from dataclasses import dataclass, field +from enum import Enum +from PySide6 import QtCore, QtNetwork import os import logging import tempfile import functools -__all__ = ['ConnectionManager'] +__all__ = ['ConnectionManager', 'ConnectionStartOperation', 'ConnectionStartStage'] logger = logging.getLogger(__name__) @@ -80,6 +89,11 @@ class _ConnectionStartAttempt: return False for runtime in reversed(self.runtimes): + setExitCallback = getattr(runtime, 'setExitCallback', None) + + if callable(setExitCallback): + setExitCallback(None) + self.manager._stopAndDisposeRuntime(runtime) self.runtimes.clear() @@ -125,6 +139,960 @@ def getUserTUNSettings(*args, **kwargs): ) +class ConnectionStartStage(Enum): + """Describe the active stage of one connection startup transaction.""" + + Pending = 'pending' + Preparing = 'preparing' + StartingPrimary = 'starting-primary' + WaitingPrimary = 'waiting-primary' + PreparingTUN = 'preparing-tun' + ResolvingTUNAddress = 'resolving-tun-address' + WaitingTUNDevice = 'waiting-tun-device' + StartingTUNRuntime = 'starting-tun-runtime' + WaitingTUNRuntime = 'waiting-tun-runtime' + ApplyingHostNetwork = 'applying-host-network' + Committing = 'committing' + Succeeded = 'succeeded' + Failed = 'failed' + Cancelled = 'cancelled' + + +class _RuntimeReadinessProbe(QtCore.QObject): + """Observe process survival and an optional local TCP endpoint.""" + + ready = QtCore.Signal() + failed = QtCore.Signal(str) + + def __init__(self, runtime, startup, parent=None): + """Initialize a bounded readiness observer.""" + super().__init__(parent) + + self._runtime = runtime + self._startup = startup + self._terminal = False + self._host = '' + self._port = 0 + + self._elapsed = QtCore.QElapsedTimer() + self._timer = QtCore.QTimer(self) + self._timer.setInterval(max(int(startup.retryInterval), 1)) + + connectWeakly(self._timer.timeout, self, '_poll') + + self._socket = QtNetwork.QTcpSocket(self) + + connectWeakly(self._socket.connected, self, '_connected') + + def start(self): + """Begin observing without blocking or nesting the Qt event loop.""" + if self._terminal or self._timer.isActive(): + return + + if self._startup.endpoint: + try: + self._host, port = parseHostPort(self._startup.endpoint) + self._port = int(port) + except Exception: + # Any non-exit exceptions + + self._finishFailed( + f'invalid runtime readiness endpoint: ' + f'{self._startup.endpoint!r}' + ) + + return + + self._elapsed.start() + self._timer.start() + self._poll() + + def _runtimeAlive(self) -> bool: + """Return whether the observed runtime still owns a live process.""" + isAlive = getattr(self._runtime, 'isAlive', None) + + return bool(callable(isAlive) and isAlive()) + + def _poll(self): + """Retry endpoint connection and enforce the startup deadline.""" + if self._terminal: + return + + if not self._runtimeAlive(): + self._finishFailed('core process exited during startup') + + return + + timeout = max(int(self._startup.timeout), 1) + + if self._elapsed.isValid() and self._elapsed.elapsed() >= timeout: + if self._startup.endpoint: + self._finishFailed('core readiness check timed out') + else: + self._finishReady() + + return + + if ( + self._startup.endpoint + and self._socket.state() + is QtNetwork.QAbstractSocket.SocketState.UnconnectedState + ): + self._socket.connectToHost(self._host, self._port) + + def _connected(self): + """Accept the first successful local endpoint connection.""" + self._finishReady() + + def _finishReady(self): + """Publish readiness exactly once.""" + if self._terminal: + return + + self._terminal = True + self._timer.stop() + self._socket.abort() + self.ready.emit() + + def _finishFailed(self, message): + """Publish startup failure exactly once.""" + if self._terminal: + return + + self._terminal = True + self._timer.stop() + self._socket.abort() + self.failed.emit(str(message)) + + def cancel(self): + """Stop all probe resources without publishing a stale result.""" + if self._terminal: + return + + self._terminal = True + self._timer.stop() + self._socket.abort() + + +class _ConditionProbe(QtCore.QObject): + """Poll one host condition through a bounded parent-owned Qt timer.""" + + finished = QtCore.Signal(bool) + + def __init__(self, predicate, description, timeout=10000, parent=None): + """Initialize an idle condition observer.""" + super().__init__(parent) + + self._predicate = predicate + self._description = description + self._timeout = max(int(timeout), 1) + self._terminal = False + + self._elapsed = QtCore.QElapsedTimer() + self._timer = QtCore.QTimer(self) + self._timer.setInterval(100) + + connectWeakly(self._timer.timeout, self, '_poll') + + def start(self): + """Begin polling the condition.""" + if self._terminal or self._timer.isActive(): + return + + self._elapsed.start() + self._timer.start() + self._poll() + + def _poll(self): + """Publish the first successful observation or one timeout.""" + if self._terminal: + return + + try: + ready = bool(self._predicate()) + except Exception as ex: + # Any non-exit exceptions + + logger.error(f'failed to inspect {self._description}: {ex}') + + ready = False + + if ready: + logger.info(f'find {self._description} success') + + self._finish(True) + elif self._elapsed.isValid() and self._elapsed.elapsed() >= self._timeout: + logger.error(f'find {self._description} failed') + + self._finish(False) + + def _finish(self, success): + """Publish exactly one terminal condition result.""" + if self._terminal: + return + + self._terminal = True + self._timer.stop() + self.finished.emit(bool(success)) + + def cancel(self): + """Stop polling without publishing a stale result.""" + if self._terminal: + return + + self._terminal = True + self._timer.stop() + + +class ConnectionStartOperation(QtCore.QObject): + """Own one cancellable, staged connection startup transaction.""" + + succeeded = QtCore.Signal(object) + failed = QtCore.Signal(object, str, str) + cancelled = QtCore.Signal(object) + stageChanged = QtCore.Signal(object) + + def __init__( + self, + manager, + generation, + config, + routing, + *, + exitCallback=None, + msgCallbackCore=None, + msgCallbackTUN_=None, + deepcopy=True, + proxyModeOnly=False, + log=True, + options=None, + parent=None, + ): + """Initialize an operation without starting any resource.""" + super().__init__(parent) + + self.manager = manager + self.generation = generation + self.routing = routing + self.exitCallback = exitCallback + self.msgCallbackCore = msgCallbackCore + self.msgCallbackTUN_ = msgCallbackTUN_ + self.proxyModeOnly = proxyModeOnly + self.log = log + self.options = dict(options or {}) + self.attempt = _ConnectionStartAttempt( + manager, manager._runtimeConfiguration(config, deepcopy) + ) + + self.stage = ConnectionStartStage.Pending + self._terminal = False + self._readinessProbe = None + self._conditionProbe = None + self._conditionContinuation = '' + self._dnsOperation = None + self._tun = None + self._startTUN = None + self._gateway = None + self._interface = None + + def _isCurrent(self): + """Return whether this generation still owns manager startup.""" + return ( + not self._terminal + and self.manager._activeStartOperation is self + and self.manager._startGeneration == self.generation + ) + + def _setStage(self, stage): + """Publish one observable startup stage.""" + if self.stage is stage: + return + + self.stage = stage + self.stageChanged.emit(stage) + + def _resume(self, methodName): + """Run one named continuation and translate exceptions into rollback.""" + if not self._isCurrent(): + return + + try: + getattr(self, methodName)() + except Exception as ex: + # Any non-exit exceptions + + self._fail('', str(ex)) + + def start(self): + """Start the semantic transaction after observers are connected.""" + if not self._isCurrent() or self.stage is not ConnectionStartStage.Pending: + return + + self._setStage(ConnectionStartStage.Preparing) + + configcopy = self.attempt.runtimeConfiguration + + try: + ( + self.attempt.nativeTUNHandled, + self.attempt.applicationTun2socks, + ) = self.manager._prepareTUNPolicy(configcopy, self.proxyModeOnly) + except TUNPreparationError as ex: + self.manager._lastStartError = str(ex) + self._fail(str(ex), 'native TUN preparation failed') + + return + except Exception as ex: + # Any non-exit exceptions + + self._fail('', str(ex)) + + return + + self._resume('_startPrimary') + + def _startPrimary(self): + """Construct and launch the primary runtime under attempt ownership.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.StartingPrimary) + + try: + launch = getPluginRegistry().createCoreRuntime( + self.attempt.runtimeConfiguration, + self.routing, + exitCallback=self.runtimeExitCallback, + messageCallback=self.msgCallbackCore, + proxyModeOnly=self.proxyModeOnly, + log=self.log, + **self.options, + ) + except Exception as ex: + # Any non-exit exceptions + + self._fail('', str(ex)) + + return + + if not isinstance(launch, CoreRuntimeLaunch): + self._fail('', 'no core runtime is available for this configuration') + + return + + runtime = launch.runtime + + self.attempt.ownRuntime(runtime) + + try: + success = ( + launch.start(waitCore=False) + if launch.startup is not None + else launch.start() + ) + except Exception as ex: + # Any non-exit exceptions + + self._fail(self._runtimeStartError(runtime), str(ex)) + + return + + if not success: + self._fail(self._runtimeStartError(runtime)) + + return + + if launch.startup is None: + self._resume('_afterPrimaryReady') + else: + self._observeRuntime( + runtime, + launch.startup, + '_afterPrimaryReady', + ConnectionStartStage.WaitingPrimary, + ) + + @staticmethod + def _runtimeStartError(runtime): + """Return one concise error published by a failed runtime.""" + startError = getattr(runtime, 'startError', None) + + return str(startError() or '') if callable(startError) else '' + + def _observeRuntime(self, runtime, startup, continuation, stage): + """Observe runtime readiness and resume through a named method.""" + if not self._isCurrent(): + return + + self._setStage(stage) + self._conditionContinuation = continuation + + probe = _RuntimeReadinessProbe(runtime, startup, parent=self) + + self._readinessProbe = probe + + connectWeakly(probe.ready, self, '_runtimeReady') + connectWeakly(probe.failed, self, '_runtimeReadinessFailed') + + probe.start() + + def _runtimeReady(self): + """Confirm the observed runtime and resume its continuation.""" + if not self._isCurrent() or self._readinessProbe is None: + return + + probe = self._readinessProbe + runtime = probe._runtime + continuation = self._conditionContinuation + + self._readinessProbe = None + self._conditionContinuation = '' + + probe.deleteLater() + + confirmStartup = getattr(runtime, 'confirmStartup', None) + + if callable(confirmStartup) and not confirmStartup(): + self._fail(self._runtimeStartError(runtime)) + + return + + self._resume(continuation) + + def _runtimeReadinessFailed(self, message): + """Fail the transaction when a readiness observer reaches terminal.""" + if not self._isCurrent(): + return + + self._readinessProbe = None + self._conditionContinuation = '' + self._fail('', message) + + def _afterPrimaryReady(self): + """Continue into application TUN or commit a proxy-only startup.""" + if not self._isCurrent(): + return + + if self.attempt.applicationTun2socks: + self._beginApplicationTun() + else: + self._commit() + + def runtimeExitCallback(self, runtime, exitcode): + """Abort when any attempt-owned runtime exits before commit.""" + if not self._isCurrent(): + return + + if exitcode == CoreRuntime.ExitCode.ConfigurationError.value: + message = 'Invalid server configuration' + elif exitcode == CoreRuntime.ExitCode.ServerStartFailure.value: + message = 'Failed to start core' + else: + try: + pluginMessage = getPluginRegistry().coreExitMessage(runtime, exitcode) + except Exception: + # Any non-exit exceptions + + pluginMessage = None + + message = pluginMessage or 'Core terminated unexpectedly' + + self._fail( + self._runtimeStartError(runtime) or message, + f'{runtime.name()} exited during startup with code {exitcode}', + ) + + def _beginApplicationTun(self): + """Acquire tun2socks and preserve the platform's startup ordering.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.PreparingTUN) + + configcopy = self.attempt.runtimeConfiguration + + if PLATFORM == 'Windows': + SystemRoutingTable.delete( + '0.0.0.0', + APPLICATION_TUN2SOCKS_GATEWAY_ADDRESS, + ) + + userGateway = userDefaultPrimaryGatewayIP() + userInterfaceIP = userPrimaryAdapterInterfaceIP() + + if userGateway and userInterfaceIP: + logger.info( + f'got user defined TUN settings. ' + f'\'DefaultPrimaryGatewayIP\': {userGateway}. ' + f'\'PrimaryAdapterInterfaceIP\': {userInterfaceIP}' + ) + self._gateway, self._interface = userGateway, userInterfaceIP + else: + logger.info( + 'automatically fetching TUN settings: ' + '\'DefaultPrimaryGatewayIP\' and ' + '\'PrimaryAdapterInterfaceIP\'' + ) + defaultGateway = SystemRoutingTable.getDefaultGateway() + + if PLATFORM == 'Darwin': + defaultGateway = list( + filter( + lambda item: (item != APPLICATION_TUN2SOCKS_GATEWAY_ADDRESS), + defaultGateway, + ) + ) + + if len(defaultGateway) != 1: + self._fail('', f'bad default gateway: {defaultGateway}') + + return + + if PLATFORM in ('Windows', 'Linux'): + self._gateway, self._interface = defaultGateway[0] + elif PLATFORM == 'Darwin': + self._gateway, self._interface = defaultGateway[0], None + else: + self._fail('', f'unrecognized platform: {PLATFORM}') + + return + + tun = Tun2socks( + exitCallback=self.runtimeExitCallback, + msgCallback=self.msgCallbackTUN_, + ) + + self._tun = tun + self.attempt.ownRuntime(tun) + + tcpSendBufferSize = userTcpSendBufferSize() + tcpReceiveBufferSize = userTcpReceiveBufferSize() + tcpAutoTuning = userTcpAutoTuning() == 'True' + interfaceArg = ( + APPLICATION_TUN2SOCKS_NETWORK_INTERFACE_NAME + if PLATFORM != 'Linux' + else self._interface + ) + + self._startTUN = functools.partial( + tun.start, + APPLICATION_TUN2SOCKS_DEVICE_NAME, + interfaceArg, + 'error', + f'socks5://{configcopy.socksProxy()}', + '', + f'{tcpSendBufferSize}MB', + f'{tcpReceiveBufferSize}MB', + tcpAutoTuning, + waitCore=False, + ) + + if PLATFORM != 'Linux': + self._setStage(ConnectionStartStage.StartingTUNRuntime) + + if not self._startTUN(): + self._fail(self._runtimeStartError(tun)) + + return + + if PLATFORM == 'Darwin': + self._observeRuntime( + tun, + CoreRuntimeStartup(), + '_prepareTunBypass', + ConnectionStartStage.WaitingTUNRuntime, + ) + else: + self._resume('_prepareTunBypass') + + def _prepareTunBypass(self): + """Prepare remote-server bypass routes without blocking for DNS.""" + if not self._isCurrent(): + return + + bypassTUN = userBypassTUNAdapterInterfaceIP() + + if bypassTUN: + for bypass in bypassTUN.split(','): + bypass = bypass.strip() + + if not isValidIPAddress(bypass): + SystemRoutingTable.managedRoutes.clear() + + self._fail( + '', + f'invalid IP address when processing user TUN ' + f'bypass settings: {bypass}', + ) + + return + + logger.info(f'processing user TUN bypass IP: {bypass}') + + SystemRoutingTable.managedRoutes.append([bypass, self._gateway]) + + self._resume('_continueTunPlatform') + + return + + address = self.attempt.runtimeConfiguration.remoteAddress() + + if isValidIPAddress(address): + SystemRoutingTable.managedRoutes.append([address, self._gateway]) + + self._resume('_continueTunPlatform') + + return + + self._setStage(ConnectionStartStage.ResolvingTUNAddress) + + resolver = self.manager._connectionDnsResolver() + resolver.configureHttpProxy(self.attempt.runtimeConfiguration.httpProxy()) + operation = resolver.resolveAsync(address, parent=self) + + self._dnsOperation = operation + + connectWeakly(operation.finished, self, '_tunAddressResolved') + + operation.start() + + def _tunAddressResolved(self, error, resolved): + """Resume TUN setup after event-driven remote-address resolution.""" + if not self._isCurrent(): + return + + operation = self._dnsOperation + + self._dnsOperation = None + + if operation is not None: + operation.deleteLater() + + if error: + SystemRoutingTable.managedRoutes.clear() + + self._fail( + '', + f'DNS resolution failed: ' + f'{self.attempt.runtimeConfiguration.remoteAddress()}', + ) + + return + + for address in resolved: + SystemRoutingTable.managedRoutes.append([address, self._gateway]) + + self._resume('_continueTunPlatform') + + def _continueTunPlatform(self): + """Continue with the platform-specific TUN stage.""" + if PLATFORM == 'Windows': + self._waitForTunDevice( + functools.partial( + SystemRoutingTable.WIN32IpconfigFindContent, + APPLICATION_TUN2SOCKS_DEVICE_NAME, + ), + '_applyWindowsTun', + ) + elif PLATFORM == 'Darwin': + self._applyDarwinTun() + elif PLATFORM == 'Linux': + self._prepareLinuxTun() + else: + self._fail('', f'unrecognized platform: {PLATFORM}') + + def _waitForTunDevice(self, predicate, continuation): + """Wait for one platform TUN device through a Qt timer.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.WaitingTUNDevice) + self._conditionContinuation = continuation + + probe = _ConditionProbe( + predicate, + f'TUN device {APPLICATION_TUN2SOCKS_DEVICE_NAME!r}', + parent=self, + ) + + self._conditionProbe = probe + + connectWeakly(probe.finished, self, '_tunDeviceObserved') + + probe.start() + + def _tunDeviceObserved(self, success): + """Resume after one device observation reaches terminal.""" + if not self._isCurrent() or self._conditionProbe is None: + return + + probe = self._conditionProbe + continuation = self._conditionContinuation + + self._conditionProbe = None + self._conditionContinuation = '' + + probe.deleteLater() + + if not success: + self._fail('', 'TUN device did not become ready') + + return + + self._resume(continuation) + + def _applyWindowsTun(self): + """Apply Windows DNS, gateway, and route mutations after readiness.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.ApplyingHostNetwork) + + userInterfaceName = userPrimaryAdapterInterfaceName() + alias = ( + userInterfaceName + if userInterfaceName + else SystemRoutingTable.WIN32GetInterfaceAliasByIP(self._interface) + ) + + if alias: + + def _windowsCleanup(_alias): + SystemRoutingTable.WIN32SetInterfaceDNS(_alias) + SystemRoutingTable.WIN32FlushDNSCache() + + self._tun.cleanup = functools.partial(_windowsCleanup, alias) + + if userDisablePrimaryAdapterInterfaceDNS() != 'False': + SystemRoutingTable.WIN32SetInterfaceDNS( + alias, + '127.0.0.1', + False, + ) + + interfaceDNS = ( + userTunAdapterInterfaceDNS() or APPLICATION_TUN2SOCKS_INTERFACE_DNS_ADDRESS + ) + + SystemRoutingTable.addRelations() + SystemRoutingTable.WIN32SetInterfaceDNS( + APPLICATION_TUN2SOCKS_DEVICE_NAME, + interfaceDNS, + False, + ) + SystemRoutingTable.setDeviceGateway( + APPLICATION_TUN2SOCKS_DEVICE_NAME, + APPLICATION_TUN2SOCKS_IP_ADDRESS, + APPLICATION_TUN2SOCKS_GATEWAY_ADDRESS, + ) + SystemRoutingTable.WIN32FlushDNSCache() + + self._commit() + + def _applyDarwinTun(self): + """Apply macOS DNS, gateway, and route mutations after startup.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.ApplyingHostNetwork) + + for address in [ + *list(f'{2 ** (8 - index)}.0.0.0/{index}' for index in range(8, 0, -1)), + '198.18.0.0/15', + ]: + SystemRoutingTable.managedRoutes.append( + [address, APPLICATION_TUN2SOCKS_GATEWAY_ADDRESS] + ) + + servers = SystemRoutingTable.DarwinGetDNSServers() + + def _darwinCleanup(_servers): + for service, dnsserver in _servers: + SystemRoutingTable.DarwinSetDNSServers(service, dnsserver) + + self._tun.cleanup = functools.partial(_darwinCleanup, servers) + + interfaceDNS = ( + userTunAdapterInterfaceDNS() or APPLICATION_TUN2SOCKS_INTERFACE_DNS_ADDRESS + ) + + for service, _dnsserver in servers: + SystemRoutingTable.DarwinSetDNSServers(service, interfaceDNS) + + SystemRoutingTable.setDeviceGateway( + APPLICATION_TUN2SOCKS_DEVICE_NAME, + APPLICATION_TUN2SOCKS_IP_ADDRESS, + APPLICATION_TUN2SOCKS_GATEWAY_ADDRESS, + ) + SystemRoutingTable.addRelations() + + self._commit() + + def _prepareLinuxTun(self): + """Create Linux TUN and routes before starting tun2socks.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.ApplyingHostNetwork) + + def _linuxCleanup(): + SystemRoutingTable.LinuxDeleteTUNDevice(APPLICATION_TUN2SOCKS_DEVICE_NAME) + + self._tun.cleanup = _linuxCleanup + + commandBringUpTUN = '' + + if not SystemRoutingTable.LinuxFindTUNDevice(APPLICATION_TUN2SOCKS_DEVICE_NAME): + commandBringUpTUN = ( + f'ip tuntap add mode tun dev ' + f'{APPLICATION_TUN2SOCKS_DEVICE_NAME}\n' + f'ip addr add 10.10.10.10/24 dev ' + f'{APPLICATION_TUN2SOCKS_DEVICE_NAME}\n' + f'ip link set dev {APPLICATION_TUN2SOCKS_DEVICE_NAME} up' + ) + + commandAddDefaultRoute = ( + f'ip route add default dev {APPLICATION_TUN2SOCKS_DEVICE_NAME} ' 'metric 5' + ) + + def route(source, destination): + return f'{source} via {destination} dev {self._interface}' + + iproute = SystemRoutingTable.LinuxGetIpRoute() + commandBypass = '\n'.join( + f'ip route add {route(sourceIP, destinationIP)}' + for sourceIP, destinationIP in SystemRoutingTable.managedRoutes + if iproute.find(route(sourceIP, destinationIP)) == -1 + ) + tempdir = os.environ.get('TMPDIR') if SystemRuntime.flatpakID() else None + + with tempfile.NamedTemporaryFile( + mode='w', + encoding='utf-8', + suffix='.sh', + dir=tempdir, + delete=True, + ) as file: + file.write( + '\n'.join( + filter( + bool, + [ + commandBringUpTUN, + commandAddDefaultRoute, + commandBypass, + ], + ) + ) + ) + file.flush() + + if not SystemRoutingTable.LinuxExecutePrivilegedScript( + file.name, + shell='bash', + ): + self._fail('', 'failed to configure Linux TUN routes') + + return + + self._waitForTunDevice( + functools.partial( + SystemRoutingTable.LinuxFindTUNDevice, + APPLICATION_TUN2SOCKS_DEVICE_NAME, + ), + '_startLinuxTun', + ) + + def _startLinuxTun(self): + """Start Linux tun2socks after the host device exists.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.StartingTUNRuntime) + + if not self._startTUN(): + self._fail(self._runtimeStartError(self._tun)) + + return + + self._observeRuntime( + self._tun, + CoreRuntimeStartup(), + '_commit', + ConnectionStartStage.WaitingTUNRuntime, + ) + + def _commit(self): + """Transfer exact runtime ownership only after every stage succeeds.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.Committing) + + for runtime in self.attempt.runtimes: + setExitCallback = getattr(runtime, 'setExitCallback', None) + + if callable(setExitCallback): + setExitCallback(self.exitCallback) + + self.attempt.commit() + self._terminal = True + self._setStage(ConnectionStartStage.Succeeded) + self.succeeded.emit(self) + self.manager._finishStartOperation(self) + self.deleteLater() + + def _cancelObservers(self): + """Cancel every child observer owned by this operation.""" + if self._readinessProbe is not None: + self._readinessProbe.cancel() + self._readinessProbe = None + + if self._conditionProbe is not None: + self._conditionProbe.cancel() + self._conditionProbe = None + + if self._dnsOperation is not None: + self._dnsOperation.cancel() + self._dnsOperation = None + + self._conditionContinuation = '' + + def _fail(self, message='', details=''): + """Roll back and publish one terminal failure.""" + if self._terminal: + return + + self._terminal = True + self._cancelObservers() + + concise = str(message or '') + + if concise: + self.manager._lastStartError = concise + + self.attempt.rollback(f'connection startup failed: {details or concise}') + self._setStage(ConnectionStartStage.Failed) + self.failed.emit(self, concise, str(details or '')) + self.manager._finishStartOperation(self) + self.deleteLater() + + def cancel(self): + """Cancel this generation and roll back only its acquired resources.""" + if self._terminal: + return False + + self._terminal = True + self._cancelObservers() + self.attempt.rollback('connection startup cancelled') + self._setStage(ConnectionStartStage.Cancelled) + self.cancelled.emit(self) + self.manager._finishStartOperation(self) + self.deleteLater() + + return True + + class ConnectionManager(Mixins.CleanupOnExit): """Coordinate proxy cores, TUN setup, DNS changes, and routing cleanup.""" @@ -137,6 +1105,8 @@ class ConnectionManager(Mixins.CleanupOnExit): self.uniqueCleanup = False self.runtimes = list() self._lastStartError = '' + self._startGeneration = 0 + self._activeStartOperation = None def _connectionDnsResolver(self) -> DnsResolver: """Return the resolver owned by this connection-manager lifecycle.""" @@ -273,7 +1243,12 @@ class ConnectionManager(Mixins.CleanupOnExit): log=True, **kwargs, ) -> bool: - """Run one staged connection attempt and roll it back on any failure.""" + """Run the legacy synchronous startup compatibility path. + + Normal GUI connections use :meth:`startAsync`. This path remains for + third-party plugins and non-interactive callers that rely on historical + synchronous launch semantics. + """ self._lastStartError = '' attempt = _ConnectionStartAttempt( @@ -298,6 +1273,56 @@ class ConnectionManager(Mixins.CleanupOnExit): raise + def startAsync( + self, + config: CoreConfiguration | ServerProfile, + routing: str, + exitCallback=None, + msgCallbackCore=None, + msgCallbackTUN_=None, + deepcopy=True, + proxyModeOnly=False, + log=True, + **kwargs, + ): + """Create and schedule one manager-owned startup transaction.""" + if self._activeStartOperation is not None: + self._activeStartOperation.cancel() + + self._lastStartError = '' + self._startGeneration += 1 + operation = ConnectionStartOperation( + self, + self._startGeneration, + config, + routing, + exitCallback=exitCallback, + msgCallbackCore=msgCallbackCore, + msgCallbackTUN_=msgCallbackTUN_, + deepcopy=deepcopy, + proxyModeOnly=proxyModeOnly, + log=log, + options=kwargs, + ) + self._activeStartOperation = operation + singleShotWeakly(0, operation, 'start') + + return operation + + def _finishStartOperation(self, operation): + """Release manager ownership of one exact terminal generation.""" + if self._activeStartOperation is operation: + self._activeStartOperation = None + + def cancelStart(self, operation=None): + """Cancel the current startup generation when it matches *operation*.""" + current = self._activeStartOperation + + if current is None or (operation is not None and current is not operation): + return False + + return current.cancel() + def _startAttempt( self, attempt: _ConnectionStartAttempt, @@ -751,11 +1776,14 @@ class ConnectionManager(Mixins.CleanupOnExit): def stopAll(self): """Stop every managed proxy-core and TUN runtime.""" + self.cancelStart() + for runtime in reversed(list(self.runtimes)): self._releaseRuntime(runtime) def cleanup(self): """Release resources owned by the core manager.""" + self.cancelStart() self.stopAll() if self._dnsResolver is not None: diff --git a/Furious/Service/DnsResolver.py b/Furious/Service/DnsResolver.py index c6cc30b..ec5f796 100644 --- a/Furious/Service/DnsResolver.py +++ b/Furious/Service/DnsResolver.py @@ -22,6 +22,7 @@ from __future__ import annotations from Furious.Frozenlib import * from Furious.Models import * from Furious.Qt.HttpGetManager import * +from Furious.Qt.Signals import connectWeakly from PySide6 import QtCore from PySide6.QtNetwork import * @@ -30,11 +31,104 @@ from typing import Tuple import logging -__all__ = ['DnsResolver'] +__all__ = ['DnsResolutionOperation', 'DnsResolver'] logger = logging.getLogger(__name__) +class DnsResolutionOperation(QtCore.QObject): + """Observe one recursive DNS request without nesting the Qt event loop.""" + + finished = QtCore.Signal(bool, object) + + def __init__(self, resolver, domain, timeout=30000, parent=None): + """Initialize an idle resolution operation.""" + super().__init__(parent) + + self._resolver = resolver + self._domain = domain + self._timeout = max(int(timeout), 1) + self._resultMap = resolver._newResultMap(domain) + self._terminal = False + + self._elapsed = QtCore.QElapsedTimer() + self._timer = QtCore.QTimer(self) + self._timer.setInterval(20) + + connectWeakly(self._timer.timeout, self, '_poll') + + def start(self): + """Start the DNS request and its event-driven completion observer.""" + if self._terminal or self._timer.isActive(): + return + + try: + self._resolver._beginResolve(self._resultMap) + except Exception as ex: + # Any non-exit exceptions + + logger.error(f'failed to start DNS resolution for {self._domain!r}: {ex}') + + self._resultMap['error'] = True + self._finish() + + return + + self._elapsed.start() + self._timer.start() + self._poll() + + def _poll(self): + """Finish when recursion drains, or abort this request at its deadline.""" + if self._terminal: + return + + if self._resultMap['depth'] == 0: + self._finish() + + return + + if self._elapsed.isValid() and self._elapsed.elapsed() >= self._timeout: + logger.error( + f'DNS resolution for {self._domain!r} reached timeout ' + f'{self._timeout // 1000}s' + ) + + self._resultMap['error'] = True + self._abortReplies() + self._finish() + + def _abortReplies(self): + """Abort only network replies acquired by this resolution.""" + for networkReply in self._resultMap['reference']: + if ( + isinstance(networkReply, QNetworkReply) + and not networkReply.isFinished() + ): + networkReply.abort() + + def _finish(self): + """Publish exactly one terminal result.""" + if self._terminal: + return + + self._terminal = True + self._timer.stop() + self.finished.emit( + bool(self._resultMap['error']), + list(self._resultMap['result'].keys()), + ) + + def cancel(self): + """Cancel without publishing a stale result.""" + if self._terminal: + return + + self._terminal = True + self._timer.stop() + self._abortReplies() + + class DnsResolver(HttpGetManager): """Represent DNS resolver.""" @@ -199,17 +293,25 @@ class DnsResolver(HttpGetManager): resultMap['error'] = True resultMap['depth'] -= 1 - def resolve(self, domain, timeout=30000) -> Tuple[bool, list[str]]: - """Resolve the DNS resolver.""" - resultMap = { + @staticmethod + def _newResultMap(domain): + """Return mutable state for one recursive DNS resolution.""" + normalizedDomain = str(domain).rstrip('.').strip().casefold() + + return { 'domain': domain, 'depth': 0, 'error': False, 'reference': [], 'result': {}, - 'visited': {str(domain).rstrip('.').strip().casefold()}, + 'visited': {normalizedDomain}, } + def _beginResolve(self, resultMap): + """Start the root request for one prepared resolution state.""" + domain = resultMap['domain'] + normalizedDomain = str(domain).rstrip('.').strip().casefold() + resultMap['depth'] += 1 networkReply = self.webGET( @@ -218,15 +320,29 @@ class DnsResolver(HttpGetManager): domain=domain, resultMap=resultMap, referenceDepth=0, - ancestry=(str(domain).rstrip('.').strip().casefold(),), + ancestry=(normalizedDomain,), ) resultMap['reference'].append(networkReply) + def resolve(self, domain, timeout=30000) -> Tuple[bool, list[str]]: + """Resolve the DNS resolver.""" + resultMap = self._newResultMap(domain) + + self._beginResolve(resultMap) self.wait(resultMap, timeout=timeout) return resultMap['error'], list(resultMap['result'].keys()) + def resolveAsync(self, domain, timeout=30000, parent=None): + """Return an event-driven DNS operation; the caller starts and owns it.""" + return DnsResolutionOperation( + self, + domain, + timeout=timeout, + parent=parent, + ) + @staticmethod def wait(resultMap, startCounter=0, timeout=30000, step=100): """Wait for the DNS resolver operation to complete.""" diff --git a/Furious/Service/__init__.py b/Furious/Service/__init__.py index 75ea5a2..2bdb8b0 100644 --- a/Furious/Service/__init__.py +++ b/Furious/Service/__init__.py @@ -19,7 +19,11 @@ from __future__ import annotations -from .ConnectionManager import ConnectionManager +from .ConnectionManager import ( + ConnectionManager, + ConnectionStartOperation, + ConnectionStartStage, +) from .ConnectivityManager import ConnectivityManager from .DnsResolver import DnsResolver from .EndpointInfoService import ( @@ -84,6 +88,8 @@ from .UpdateManager import UpdateManager __all__ = [ 'ConnectionManager', + 'ConnectionStartOperation', + 'ConnectionStartStage', 'ConnectivityManager', 'DnsResolver', 'EndpointInfo', diff --git a/tests/README.md b/tests/README.md index 34a932c..25693db 100644 --- a/tests/README.md +++ b/tests/README.md @@ -40,7 +40,7 @@ strategy in an individual test. | Configuration, profiles, migration, repositories | `test_models_and_services.py`, `test_repository_contracts.py` | | Generation log invariants, model fuzzing, concurrency, reclamation, complexity, and opt-in soak/latency probes | `test_log_manager_generation.py` | | Low-level application, runtime, editor, and storage contracts | `test_interface.py` | -| Application composition, startup rollback, connection ownership, entry-point and crash boundaries | `test_architecture_refactors.py`, `test_application_process.py` | +| Application composition, asynchronous readiness/TUN startup, rollback, connection ownership, entry-point and crash boundaries | `test_architecture_refactors.py`, `test_connection_startup_async.py`, `test_application_process.py` | | Plugin registration, capability dispatch, factories, rollback, and Hysteria1 ownership | `test_plugin_architecture.py`, `test_hysteria1_protocol.py` | | Controller state and error transitions with injected runtimes | `test_controllers.py` | | SOCKS and SIP002 Shadowsocks codecs and generated round trips | `test_socks_uri.py`, `test_shadowsocks_uri.py` | @@ -109,7 +109,7 @@ Then run the desired test tier. python -m unittest discover -s tests -v # Regular logic, persistence, plugin, controller, codec, and UI regressions -python -m unittest tests.test_interface tests.test_models_and_services tests.test_repository_contracts tests.test_architecture_refactors tests.test_plugin_architecture tests.test_hysteria1_protocol tests.test_hysteria2_compatibility tests.test_controllers tests.test_subscription_manager tests.test_subscription_sync tests.test_profile_test_jobs tests.test_socks_uri tests.test_shadowsocks_uri tests.test_backend_editor_contract tests.test_xray_asset_download tests.test_native_tun_semantics tests.test_metrics_behavior tests.test_endpoint_info tests.test_service_runtime tests.test_frozenlib tests.test_isolation_and_navigation tests.test_main_window_geometry tests.test_dialog_geometry tests.test_ui_behavior tests.test_qt_interactions tests.test_stylesheet_states tests.test_theme_transition tests.test_public_api -v +python -m unittest tests.test_interface tests.test_models_and_services tests.test_repository_contracts tests.test_architecture_refactors tests.test_connection_startup_async tests.test_plugin_architecture tests.test_hysteria1_protocol tests.test_hysteria2_compatibility tests.test_controllers tests.test_subscription_manager tests.test_subscription_sync tests.test_profile_test_jobs tests.test_socks_uri tests.test_shadowsocks_uri tests.test_backend_editor_contract tests.test_xray_asset_download tests.test_native_tun_semantics tests.test_metrics_behavior tests.test_endpoint_info tests.test_service_runtime tests.test_frozenlib tests.test_isolation_and_navigation tests.test_main_window_geometry tests.test_dialog_geometry tests.test_ui_behavior tests.test_qt_interactions tests.test_stylesheet_states tests.test_theme_transition tests.test_public_api -v # Direct Qt/process integration and destruction/lifetime checks python -m unittest tests.test_application_process tests.test_external_core tests.test_layout_matrix tests.test_qt_lifetime -v @@ -126,7 +126,7 @@ python -m unittest tests.test_very_heavy -v python -m unittest tests.test_log_manager_generation.VeryHeavyGenerationLogManagerTest -v # Shared-state order-independence spot check -python -m unittest tests.test_public_api tests.test_theme_transition tests.test_stylesheet_states tests.test_qt_interactions tests.test_ui_behavior tests.test_dialog_geometry tests.test_main_window_geometry tests.test_isolation_and_navigation tests.test_frozenlib tests.test_service_runtime tests.test_endpoint_info tests.test_metrics_behavior tests.test_native_tun_semantics tests.test_xray_asset_download tests.test_backend_editor_contract tests.test_shadowsocks_uri tests.test_socks_uri tests.test_profile_test_jobs tests.test_subscription_sync tests.test_subscription_manager tests.test_controllers tests.test_hysteria2_compatibility tests.test_hysteria1_protocol tests.test_plugin_architecture tests.test_architecture_refactors tests.test_repository_contracts tests.test_models_and_services tests.test_interface -v +python -m unittest tests.test_public_api tests.test_theme_transition tests.test_stylesheet_states tests.test_qt_interactions tests.test_ui_behavior tests.test_dialog_geometry tests.test_main_window_geometry tests.test_isolation_and_navigation tests.test_frozenlib tests.test_service_runtime tests.test_endpoint_info tests.test_metrics_behavior tests.test_native_tun_semantics tests.test_xray_asset_download tests.test_backend_editor_contract tests.test_shadowsocks_uri tests.test_socks_uri tests.test_profile_test_jobs tests.test_subscription_sync tests.test_subscription_manager tests.test_controllers tests.test_hysteria2_compatibility tests.test_hysteria1_protocol tests.test_plugin_architecture tests.test_connection_startup_async tests.test_architecture_refactors tests.test_repository_contracts tests.test_models_and_services tests.test_interface -v python -m unittest discover -s tests -v ``` diff --git a/tests/test_connection_startup_async.py b/tests/test_connection_startup_async.py new file mode 100644 index 0000000..40b5480 --- /dev/null +++ b/tests/test_connection_startup_async.py @@ -0,0 +1,786 @@ +# 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 event-driven connection startup with real Qt event delivery.""" + +from __future__ import annotations + +from Furious.Interface import CoreRuntime +from Furious.Plugins import CoreRuntimeLaunch, CoreRuntimeStartup +from Furious.Qt.Signals import singleShotWeakly +from Furious.Service.ConnectionManager import ( + ConnectionManager, + ConnectionStartStage, +) +from Furious.Service.DnsResolver import DnsResolutionOperation + +from PySide6 import QtCore, QtNetwork + +from tests.support import application, processQtEvents, waitFor + +import importlib +import unittest + +from unittest import TestCase, mock + + +class _Configuration: + """Provide the endpoint values used by manager-only tests.""" + + def httpProxy(self): + """Return a deterministic local proxy endpoint.""" + return '127.0.0.1:18080' + + def socksProxy(self): + """Return a deterministic local SOCKS endpoint.""" + return '127.0.0.1:18081' + + def remoteAddress(self): + """Avoid DNS during mocked TUN tests.""" + return '192.0.2.1' + + +class _Runtime(CoreRuntime): + """Provide an observable in-process stand-in for a core runtime.""" + + def __init__(self, exitCallback=None): + """Initialize an idle runtime and lifecycle counters.""" + super().__init__(exitCallback) + + self.alive = False + self.startOptions = [] + self.stopCount = 0 + self.disposeCount = 0 + self.confirmCount = 0 + + @staticmethod + def name(): + """Return the fixture runtime name.""" + return 'Async Fixture' + + @staticmethod + def version(): + """Return a fixture version.""" + return '1.0' + + def start(self, _configuration=None, *_args, **kwargs): + """Become live and retain launch options.""" + self.startOptions.append(dict(kwargs)) + self.alive = True + + return True + + def stop(self): + """Stop this exact runtime.""" + self.stopCount += 1 + self.alive = False + + def dispose(self): + """Record final resource disposal.""" + self.disposeCount += 1 + self._exitCallback = None + + def isAlive(self): + """Return the controlled process-survival state.""" + return self.alive + + def confirmStartup(self): + """Record readiness confirmation.""" + if not self.alive: + return False + + self.confirmCount += 1 + + return True + + def fail(self, exitcode=61): + """Simulate an early child-process exit.""" + self.alive = False + self.callExitCallback(exitcode) + + +class _Registry: + """Return prepared launches in deterministic order.""" + + def __init__(self, launches): + """Retain a launch queue.""" + self.launches = list(launches) + + def createCoreRuntime(self, *_args, **_kwargs): + """Return the next prepared runtime launch.""" + return self.launches.pop(0) + + +class _ResolverFixture(QtCore.QObject): + """Complete one recursive-resolution state through a real Qt timer.""" + + def __init__(self): + """Initialize without a pending result.""" + super().__init__() + + self.resultMap = None + + @staticmethod + def _newResultMap(domain): + """Return the minimum state required by the observer.""" + return { + 'domain': domain, + 'depth': 0, + 'error': False, + 'reference': [], + 'result': {}, + } + + def _beginResolve(self, resultMap): + """Schedule asynchronous completion.""" + self.resultMap = resultMap + resultMap['depth'] = 1 + singleShotWeakly(0, self, '_complete') + + def _complete(self): + """Publish one deterministic address through shared state.""" + self.resultMap['result']['192.0.2.10'] = True + self.resultMap['depth'] = 0 + + +class ConnectionStartupAsyncTest(TestCase): + """Verify readiness, cancellation, rollback, and compatibility.""" + + def setUp(self): + """Ensure a Qt application exists for real timer/socket delivery.""" + self.app = application() + self.servers = [] + self.managers = [] + + def tearDown(self): + """Release every exact manager and local listener.""" + for manager in self.managers: + manager.cleanup() + + for server in self.servers: + server.close() + server.deleteLater() + + processQtEvents() + + def _manager(self): + """Return a manager with host TUN disabled at the test boundary.""" + manager = ConnectionManager() + manager._prepareTUNPolicy = mock.Mock(return_value=(False, False)) + self.managers.append(manager) + + return manager + + def _server(self): + """Listen on one real loopback endpoint.""" + server = QtNetwork.QTcpServer(self.app) + self.assertTrue( + server.listen( + QtNetwork.QHostAddress.SpecialAddress.LocalHost, + 0, + ) + ) + self.servers.append(server) + + return server + + def _operation(self, manager, launch): + """Start one operation through an injected registry.""" + registry = _Registry([launch]) + patcher = mock.patch( + 'Furious.Service.ConnectionManager.getPluginRegistry', + return_value=registry, + ) + patcher.start() + self.addCleanup(patcher.stop) + + return manager.startAsync( + _Configuration(), + 'Global', + deepcopy=False, + ) + + def testLocalEndpointReadinessCompletesEarlyAndKeepsQtResponsive(self): + """Commit as soon as a real listener accepts while timers still run.""" + server = self._server() + endpoint = f'127.0.0.1:{server.serverPort()}' + runtime = _Runtime() + launch = CoreRuntimeLaunch( + runtime, + _Configuration(), + startup=CoreRuntimeStartup(endpoint=endpoint, timeout=1500), + ) + manager = self._manager() + operation = self._operation(manager, launch) + succeeded = [] + timerEvents = [] + operation.succeeded.connect(succeeded.append) + QtCore.QTimer.singleShot(0, lambda: timerEvents.append(True)) + + self.assertEqual(operation.stage, ConnectionStartStage.Pending) + self.assertTrue(waitFor(lambda: bool(succeeded), timeout=0.5)) + self.assertEqual(timerEvents, [True]) + self.assertEqual(manager.runtimes, [runtime]) + self.assertEqual(runtime.confirmCount, 1) + self.assertFalse(runtime.startOptions[0]['waitCore']) + + def testReadinessTimeoutRollsBackTheExactRuntime(self): + """Fail a live process whose promised local endpoint never appears.""" + server = self._server() + port = server.serverPort() + server.close() + runtime = _Runtime() + manager = self._manager() + operation = self._operation( + manager, + CoreRuntimeLaunch( + runtime, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint=f'127.0.0.1:{port}', + timeout=60, + retryInterval=5, + ), + ), + ) + failures = [] + operation.failed.connect(lambda *_args: failures.append(_args)) + + self.assertTrue(waitFor(lambda: bool(failures))) + self.assertEqual(manager.runtimes, []) + self.assertEqual(runtime.stopCount, 1) + self.assertEqual(runtime.disposeCount, 1) + + def testEarlyRuntimeExitFailsOnceAndIgnoresLateProbeEvents(self): + """Let process termination own the terminal result before timeout.""" + runtime = _Runtime() + manager = self._manager() + operation = self._operation( + manager, + CoreRuntimeLaunch( + runtime, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint='127.0.0.1:1', + timeout=500, + ), + ), + ) + failures = [] + operation.failed.connect(lambda *_args: failures.append(_args)) + + processQtEvents() + singleShotWeakly(0, runtime, 'fail') + + self.assertTrue(waitFor(lambda: bool(failures))) + processQtEvents(5) + self.assertEqual(len(failures), 1) + self.assertEqual(runtime.stopCount, 1) + self.assertEqual(runtime.disposeCount, 1) + + def testCancellationAndReplacementCannotCommitAStaleGeneration(self): + """Cancel the first generation before a second ready launch commits.""" + server = self._server() + first = _Runtime() + second = _Runtime() + manager = self._manager() + registry = _Registry( + [ + CoreRuntimeLaunch( + first, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint='127.0.0.1:1', + timeout=5000, + ), + ), + CoreRuntimeLaunch( + second, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint=f'127.0.0.1:{server.serverPort()}', + timeout=500, + ), + ), + ] + ) + + with mock.patch( + 'Furious.Service.ConnectionManager.getPluginRegistry', + return_value=registry, + ): + firstOperation = manager.startAsync( + _Configuration(), + 'Global', + deepcopy=False, + ) + cancelled = [] + firstOperation.cancelled.connect(cancelled.append) + processQtEvents() + + secondOperation = manager.startAsync( + _Configuration(), + 'Global', + deepcopy=False, + ) + succeeded = [] + secondOperation.succeeded.connect(succeeded.append) + + self.assertTrue(waitFor(lambda: bool(succeeded))) + + self.assertEqual(cancelled, [firstOperation]) + self.assertEqual(first.stopCount, 1) + self.assertEqual(first.disposeCount, 1) + self.assertEqual(manager.runtimes, [second]) + + def testLegacyLaunchRetainsSynchronousStartOptions(self): + """Do not inject waitCore into a plugin without async capability.""" + runtime = _Runtime() + manager = self._manager() + operation = self._operation( + manager, + CoreRuntimeLaunch(runtime, _Configuration()), + ) + succeeded = [] + operation.succeeded.connect(succeeded.append) + + self.assertTrue(waitFor(lambda: bool(succeeded))) + self.assertEqual(runtime.startOptions, [{}]) + self.assertEqual(runtime.confirmCount, 0) + self.assertEqual(manager.runtimes, [runtime]) + + def testDnsResolutionOperationCompletesAndCancelsWithoutNestedWait(self): + """Observe recursive DNS state through timers and suppress stale cancel.""" + resolver = _ResolverFixture() + operation = DnsResolutionOperation( + resolver, + 'example.test', + timeout=200, + ) + results = [] + operation.finished.connect( + lambda error, addresses: results.append((error, addresses)) + ) + operation.start() + + self.assertTrue(waitFor(lambda: bool(results))) + self.assertEqual(results, [(False, ['192.0.2.10'])]) + + cancelled = DnsResolutionOperation( + resolver, + 'cancelled.test', + timeout=200, + ) + staleResults = [] + cancelled.finished.connect( + lambda error, addresses: staleResults.append((error, addresses)) + ) + cancelled.start() + cancelled.cancel() + processQtEvents(5) + + self.assertEqual(staleResults, []) + operation.deleteLater() + cancelled.deleteLater() + resolver.deleteLater() + + def testLinuxTunPreservesDeviceBeforeRuntimeOrderingWithoutNestedWait(self): + """Create and observe the Linux device before launching tun2socks.""" + module = importlib.import_module('Furious.Service.ConnectionManager') + server = self._server() + primary = _Runtime() + tun = _Runtime() + tun.cleanup = None + events = [] + deviceChecks = 0 + + def tunFactory(**kwargs): + tun.setExitCallback(kwargs.get('exitCallback')) + + return tun + + originalTunStart = tun.start + + def startTun(*args, **kwargs): + events.append('tun-start') + + return originalTunStart(*args, **kwargs) + + tun.start = startTun + + def findDevice(_name): + nonlocal deviceChecks + + deviceChecks += 1 + events.append('find-device') + + return deviceChecks >= 2 + + def executeScript(*_args, **_kwargs): + events.append('script') + + return True + + manager = ConnectionManager() + manager._prepareTUNPolicy = mock.Mock(return_value=(False, True)) + self.managers.append(manager) + registry = _Registry( + [ + CoreRuntimeLaunch( + primary, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint=f'127.0.0.1:{server.serverPort()}', + timeout=500, + ), + ) + ] + ) + shortSurvival = CoreRuntimeStartup(timeout=20, retryInterval=5) + + with ( + mock.patch.object(module, 'PLATFORM', 'Linux'), + mock.patch.object(module, 'Tun2socks', side_effect=tunFactory), + mock.patch.object( + module, + 'CoreRuntimeStartup', + return_value=shortSurvival, + ), + mock.patch.object( + module, + 'getPluginRegistry', + return_value=registry, + ), + mock.patch.object( + module.SystemRoutingTable, + 'managedRoutes', + [], + ), + mock.patch.object( + module.SystemRoutingTable, + 'getDefaultGateway', + return_value=[('192.0.2.254', 'eth0')], + ), + mock.patch.object( + module.SystemRoutingTable, + 'LinuxFindTUNDevice', + side_effect=findDevice, + ), + mock.patch.object( + module.SystemRoutingTable, + 'LinuxGetIpRoute', + return_value='', + ), + mock.patch.object( + module.SystemRoutingTable, + 'LinuxExecutePrivilegedScript', + side_effect=executeScript, + ), + mock.patch.object( + module.SystemRoutingTable, + 'LinuxDeleteTUNDevice', + ), + mock.patch.object( + module.SystemRoutingTable, + 'deleteRelations', + ), + mock.patch.object( + module.SystemRuntime, + 'flatpakID', + return_value='', + ), + mock.patch.object( + module.PySide6Legacy, + 'eventLoopWait', + side_effect=AssertionError('nested event loop used'), + ), + ): + operation = manager.startAsync( + _Configuration(), + 'Global', + deepcopy=False, + ) + succeeded = [] + operation.succeeded.connect(succeeded.append) + + self.assertTrue(waitFor(lambda: bool(succeeded))) + + self.assertEqual( + events[:4], + ['find-device', 'script', 'find-device', 'tun-start'], + ) + self.assertEqual(manager.runtimes, [primary, tun]) + self.assertFalse(tun.startOptions[0]['waitCore']) + + def testWindowsTunStartsRuntimeBeforeObservingAndMutatingDevice(self): + """Keep the Windows launch, device, then host-mutation sequence.""" + module = importlib.import_module('Furious.Service.ConnectionManager') + server = self._server() + primary = _Runtime() + tun = _Runtime() + tun.cleanup = None + events = [] + + def tunFactory(**kwargs): + tun.setExitCallback(kwargs.get('exitCallback')) + + return tun + + originalTunStart = tun.start + + def startTun(*args, **kwargs): + events.append('tun-start') + + return originalTunStart(*args, **kwargs) + + tun.start = startTun + + def findDevice(_name): + events.append('find-device') + + return True + + def addRelations(): + events.append('add-relations') + + manager = ConnectionManager() + manager._prepareTUNPolicy = mock.Mock(return_value=(False, True)) + self.managers.append(manager) + registry = _Registry( + [ + CoreRuntimeLaunch( + primary, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint=f'127.0.0.1:{server.serverPort()}', + timeout=500, + ), + ) + ] + ) + + with ( + mock.patch.object(module, 'PLATFORM', 'Windows'), + mock.patch.object(module, 'Tun2socks', side_effect=tunFactory), + mock.patch.object( + module, + 'getPluginRegistry', + return_value=registry, + ), + mock.patch.object( + module.SystemRoutingTable, + 'managedRoutes', + [], + ), + mock.patch.object( + module.SystemRoutingTable, + 'delete', + ), + mock.patch.object( + module.SystemRoutingTable, + 'getDefaultGateway', + return_value=[('192.0.2.254', '192.0.2.10')], + ), + mock.patch.object( + module.SystemRoutingTable, + 'WIN32IpconfigFindContent', + side_effect=findDevice, + ), + mock.patch.object( + module.SystemRoutingTable, + 'WIN32GetInterfaceAliasByIP', + return_value='Ethernet', + ), + mock.patch.object( + module.SystemRoutingTable, + 'WIN32SetInterfaceDNS', + ), + mock.patch.object( + module.SystemRoutingTable, + 'WIN32FlushDNSCache', + ), + mock.patch.object( + module.SystemRoutingTable, + 'setDeviceGateway', + ), + mock.patch.object( + module.SystemRoutingTable, + 'addRelations', + side_effect=addRelations, + ), + mock.patch.object( + module.SystemRoutingTable, + 'deleteRelations', + ), + mock.patch.object( + module.PySide6Legacy, + 'eventLoopWait', + side_effect=AssertionError('nested event loop used'), + ), + ): + operation = manager.startAsync( + _Configuration(), + 'Global', + deepcopy=False, + ) + succeeded = [] + operation.succeeded.connect(succeeded.append) + + self.assertTrue(waitFor(lambda: bool(succeeded))) + + self.assertEqual( + events[:3], + ['tun-start', 'find-device', 'add-relations'], + ) + self.assertEqual(manager.runtimes, [primary, tun]) + + def testDarwinTunSurvivalPrecedesDnsAndRouteMutation(self): + """Observe tun2socks survival before applying macOS host networking.""" + module = importlib.import_module('Furious.Service.ConnectionManager') + server = self._server() + primary = _Runtime() + tun = _Runtime() + tun.cleanup = None + events = [] + + def tunFactory(**kwargs): + tun.setExitCallback(kwargs.get('exitCallback')) + + return tun + + originalTunStart = tun.start + + def startTun(*args, **kwargs): + events.append('tun-start') + + return originalTunStart(*args, **kwargs) + + tun.start = startTun + + def dnsServers(): + events.append('read-dns') + + return [('Wi-Fi', ['192.0.2.53'])] + + manager = ConnectionManager() + manager._prepareTUNPolicy = mock.Mock(return_value=(False, True)) + self.managers.append(manager) + registry = _Registry( + [ + CoreRuntimeLaunch( + primary, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint=f'127.0.0.1:{server.serverPort()}', + timeout=500, + ), + ) + ] + ) + shortSurvival = CoreRuntimeStartup(timeout=20, retryInterval=5) + + with ( + mock.patch.object(module, 'PLATFORM', 'Darwin'), + mock.patch.object(module, 'Tun2socks', side_effect=tunFactory), + mock.patch.object( + module, + 'CoreRuntimeStartup', + return_value=shortSurvival, + ), + mock.patch.object( + module, + 'getPluginRegistry', + return_value=registry, + ), + mock.patch.object( + module.SystemRoutingTable, + 'managedRoutes', + [], + ), + mock.patch.object( + module.SystemRoutingTable, + 'getDefaultGateway', + return_value=['192.0.2.254'], + ), + mock.patch.object( + module.SystemRoutingTable, + 'DarwinGetDNSServers', + side_effect=dnsServers, + ), + mock.patch.object( + module.SystemRoutingTable, + 'DarwinSetDNSServers', + ), + mock.patch.object( + module.SystemRoutingTable, + 'setDeviceGateway', + ), + mock.patch.object( + module.SystemRoutingTable, + 'addRelations', + ), + mock.patch.object( + module.SystemRoutingTable, + 'deleteRelations', + ), + mock.patch.object( + module.PySide6Legacy, + 'eventLoopWait', + side_effect=AssertionError('nested event loop used'), + ), + ): + operation = manager.startAsync( + _Configuration(), + 'Global', + deepcopy=False, + ) + succeeded = [] + operation.succeeded.connect(succeeded.append) + + self.assertTrue(waitFor(lambda: bool(succeeded))) + + self.assertEqual(events, ['tun-start', 'read-dns']) + self.assertEqual(manager.runtimes, [primary, tun]) + + def testRepeatedCancellationReleasesOperationTimersAndRuntimes(self): + """Keep repeated startup cancellation bounded and independently owned.""" + manager = self._manager() + + for _index in range(12): + runtime = _Runtime() + operation = self._operation( + manager, + CoreRuntimeLaunch( + runtime, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint='127.0.0.1:1', + timeout=5000, + ), + ), + ) + processQtEvents() + self.assertTrue(manager.cancelStart(operation)) + processQtEvents() + self.assertEqual(runtime.stopCount, 1) + self.assertEqual(runtime.disposeCount, 1) + self.assertIsNone(manager._activeStartOperation) + + +if __name__ == '__main__': + application() + unittest.main() diff --git a/tests/test_controllers.py b/tests/test_controllers.py index 5e2d0f6..73745ac 100644 --- a/tests/test_controllers.py +++ b/tests/test_controllers.py @@ -29,6 +29,8 @@ from Furious.Frozenlib import AppBinarySettings, AppSettings from Furious.Models import CoreConfiguration, ServerProfile from Furious.Service.LogManager import LogManager +from PySide6 import QtCore + from tests.support import application, isolatedSettings, processQtEvents import unittest @@ -87,6 +89,53 @@ class FixtureCoreManager: raise self.stopException +class FixtureStartOperation(QtCore.QObject): + """Publish controllable asynchronous startup outcomes.""" + + succeeded = QtCore.Signal(object) + failed = QtCore.Signal(object, str, str) + cancelled = QtCore.Signal(object) + + def succeed(self): + """Publish one successful manager commit.""" + self.succeeded.emit(self) + + def fail(self, message='fixture failure', details=''): + """Publish one failed manager transaction.""" + self.failed.emit(self, message, details) + + +class FixtureAsyncCoreManager: + """Record controller use of the asynchronous manager boundary.""" + + def __init__(self): + """Initialize empty operation and runtime history.""" + self.lastStartError = '' + self.runtimes = [] + self.operations = [] + self.cancelCalls = [] + self.stopCalls = 0 + + def startAsync(self, configuration, **kwargs): + """Return one idle operation for the controller to observe.""" + operation = FixtureStartOperation() + self.operations.append((operation, configuration, kwargs)) + + return operation + + def cancelStart(self, operation): + """Cancel one exact operation.""" + self.cancelCalls.append(operation) + operation.cancelled.emit(operation) + + return True + + def stopAll(self): + """Record stable-runtime cleanup.""" + self.stopCalls += 1 + self.runtimes.clear() + + class FixtureUpdatesManager: """Record update hooks without contacting any update service.""" @@ -313,6 +362,103 @@ class ConnectionControllerTest(unittest.TestCase): controller.deleteLater() + def testAsyncManagerCommitsSystemProxyOnlyAfterSuccess(self): + """Remain Connecting until the manager transaction commits.""" + with isolatedSettings(): + core = FixtureAsyncCoreManager() + controller = ConnectionController( + coreManager=core, + updatesManager=FixtureUpdatesManager(), + ) + + with ( + mock.patch( + 'Furious.Controllers.ConnectionController.SystemProxy.set' + ) as proxySet, + mock.patch('Furious.Controllers.ConnectionController.SystemProxy.off'), + mock.patch.object(controller, '_runPostConnectTasksOnce'), + ): + self.assertTrue(controller.startConnection(self.profile)) + self.assertTrue(controller.isConnecting()) + proxySet.assert_not_called() + + operation = core.operations[0][0] + core.runtimes.append(object()) + operation.succeed() + + self.assertTrue(controller.isConnected()) + proxySet.assert_called_once() + self.assertTrue(controller.startDisconnection()) + + controller.deleteLater() + + def testDisconnectCancelsAnInFlightStartupGeneration(self): + """Cancel partial startup without ever enabling the system proxy.""" + with isolatedSettings(): + core = FixtureAsyncCoreManager() + controller = ConnectionController( + coreManager=core, + updatesManager=FixtureUpdatesManager(), + ) + + with ( + mock.patch( + 'Furious.Controllers.ConnectionController.SystemProxy.set' + ) as proxySet, + mock.patch('Furious.Controllers.ConnectionController.SystemProxy.off'), + ): + self.assertTrue(controller.startConnection(self.profile)) + operation = core.operations[0][0] + self.assertTrue(controller.startDisconnection()) + + self.assertEqual(core.cancelCalls, [operation]) + self.assertTrue(controller.state is ConnectionState.Disconnected) + proxySet.assert_not_called() + + controller.deleteLater() + + def testReconnectCancelsConnectingGenerationBeforeReplacement(self): + """Replace a Connecting attempt without accepting stale completion.""" + with isolatedSettings(): + core = FixtureAsyncCoreManager() + controller = ConnectionController( + coreManager=core, + updatesManager=FixtureUpdatesManager(), + ) + + with ( + mock.patch( + 'Furious.Controllers.ConnectionController.Storage.UserServers', + return_value=[self.profile], + ), + mock.patch( + 'Furious.Controllers.ConnectionController.Storage.UserActivatedItemIndex', + return_value=0, + ), + mock.patch( + 'Furious.Controllers.ConnectionController.SystemProxy.set' + ) as proxySet, + mock.patch('Furious.Controllers.ConnectionController.SystemProxy.off'), + ): + self.assertTrue(controller.startConnection(self.profile)) + first = core.operations[0][0] + self.assertTrue(controller.startReconnection()) + second = core.operations[1][0] + + first.succeed() + self.assertTrue(controller.isConnecting()) + proxySet.assert_not_called() + + core.runtimes.append(object()) + second.succeed() + self.assertTrue(controller.isConnected()) + proxySet.assert_called_once() + controller.startDisconnection() + + self.assertEqual(core.cancelCalls, [first]) + + controller.deleteLater() + def testStartAndProxyExceptionsReturnToStableDisconnectedState(self): """Clean every partially acquired resource after injected failures.""" for core, proxySideEffect in (