Expand test coverage and fix lifecycle contracts

Signed-off-by: Loren Eteval <loren.eteval@proton.me>
This commit is contained in:
Loren Eteval
2026-08-23 08:55:24 +08:00
parent 68ad1eefc1
commit 185bc784cb
13 changed files with 1408 additions and 17 deletions
+1 -1
View File
@@ -163,7 +163,7 @@ class Storage:
)
prefix = f'{index + 1} - ' if index >= 0 else ''
return prefix + configuration.itemRemark
return prefix + profile.itemRemark
except Exception:
# Any non-exit exceptions
+7 -2
View File
@@ -38,13 +38,18 @@ class UserTUNSettings(Mixins.CleanupOnExit, StorageBackend):
def restore():
"""Restore the user TUN settings."""
try:
return UJSONEncoder.decode(
data = UJSONEncoder.decode(
PyBase64Encoder.decode(AppSettings.get('CustomTUNSettings'))
)
if isinstance(data, dict):
return data
except Exception:
# Any non-exit exceptions
return {}
pass
return {}
self._data = restore()
+7 -1
View File
@@ -53,7 +53,10 @@ class AppMainProcess(ProcessContext.Process):
self.startupTime = str(datetime.datetime.now()).replace(':', '')
self.logFileName = f'{self.startupTime}.log'
self.fileWritten = multiprocessing.Manager().Value('b', False)
# One synchronized scalar is sufficient for the parent/child result.
# A multiprocessing.Manager would create a second child process whose
# lifecycle is unrelated to the exact application process we own.
self.fileWritten = ProcessContext.Value('b', False)
self.func = func
self.application = None
@@ -120,6 +123,9 @@ class AppMainProcess(ProcessContext.Process):
"""Handle r."""
logger.info(f'received signal {signal.Signals(signum).name}. Exit application')
if self.application is None:
sys.exit(ApplicationRunner.ExitCode.ExitSuccess.value)
self.application.exit()
def run(self):
+5
View File
@@ -20,6 +20,11 @@
TUN, and proxy-only stripping must not share a helper that erases their differences.
- Keep deterministic logic/UI behavior in the normal tier. Put repeated live-object, native-handle, thread, subprocess,
and RSS trends in explicit stress tiers.
- Scale cheap fake/model operations into the thousands where useful. Keep real Qt, WebEngine, and subprocess counts
proportional to their cost; workloads beyond the normal stress tier require the explicit
`FURIOUS_VERY_HEAVY_TESTS=1` opt-in.
- Warm up allocators before interpreting RSS, sample only at bounded batch boundaries, and prefer exact live-object,
registry, thread, handle, and child-process ownership assertions over arbitrary memory thresholds.
- Qt lifetime tests use real close/deferred-delete paths, weak references, destroyed signals, and live counts.
`gc.collect()` is diagnostic at batch boundaries only; threshold increases or production collection are not leak
fixes.
+21 -13
View File
@@ -2,8 +2,8 @@
The suite uses Python's built-in `unittest` runner. In the main test process,
Qt tests use one deliberately small `QApplication` and route `QSettings` to a
unique temporary directory. Focused lifecycle regressions may start a hermetic
child Python process, and external-runtime tests use that child interpreter as
unique temporary directory. Focused lifecycle, public-import, and display-scale
regressions may start a hermetic child Python process, and external-runtime tests use that child interpreter as
a harmless stand-in for a core. Tests do not initialize Furious's singleton IPC
server, production repositories, system proxy, TUN, routing, update network
clients, or real proxy cores.
@@ -12,10 +12,10 @@ clients, or real proxy cores.
| Area | Principal tests |
| --- | --- |
| Configuration, profiles, migration, repositories | `test_models_and_services.py` |
| Configuration, profiles, migration, repositories | `test_models_and_services.py`, `test_repository_contracts.py` |
| Low-level application, runtime, editor, and storage contracts | `test_interface.py` |
| Application composition, startup rollback, and connection ownership | `test_architecture_refactors.py` |
| Plugin registration, capability dispatch, factories, rollback | `test_plugin_architecture.py` |
| Application composition, startup rollback, connection ownership, entry-point and crash boundaries | `test_architecture_refactors.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` |
| Subscription workflow, timers, stale requests, and reconciliation | `test_subscription_manager.py`, `test_subscription_sync.py` |
@@ -26,13 +26,14 @@ clients, or real proxy cores.
| Proxy-only endpoint discovery, caching, and presentation | `test_endpoint_info.py` |
| Bounded service work, update validation, plugin UI, and worker lifetime | `test_service_runtime.py` |
| Frozenlib state helpers and mocked platform-operation boundaries | `test_frozenlib.py` |
| Settings sandbox and navigation overlay behavior | `test_isolation_and_navigation.py` |
| Settings sandbox, navigation overlay behavior, public exports, and scale/theme isolation | `test_isolation_and_navigation.py`, `test_public_api.py`, `test_layout_matrix.py` |
| AppQMainWindow lifecycle, subclass policies, geometry restoration, and migration | `test_main_window_geometry.py` |
| AppQDialog first-presentation geometry, native show paths, centering, and async lifetime | `test_dialog_geometry.py` |
| Editor mappings, lazy log rendering, routing/message-box/connection UI | `test_ui_behavior.py` |
| Direct Qt ownership and destruction across independent UI families | `test_qt_lifetime.py` |
| Batched real/probe Qt object, handle, Python allocation, and RSS trends | `test_qt_stress.py` |
| Repeated harmless subprocess, pipe, thread, handle, and RSS trends | `test_process_stress.py` |
| Opt-in release-confidence counts (100 app children, 100 external cores, 100k metrics, 40k logs, 20k navigation, 5k plugins, 1k dialogs) | `test_very_heavy.py` |
The lifecycle tests classify `AppQTransientDialog`, protocol/plugin editors,
routing dialogs, subscription editors, message boxes, QR windows, and TUN
@@ -78,16 +79,21 @@ 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_architecture_refactors tests.test_plugin_architecture tests.test_controllers tests.test_subscription_manager tests.test_subscription_sync tests.test_socks_uri tests.test_shadowsocks_uri tests.test_backend_editor_contract 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 -v
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_socks_uri tests.test_shadowsocks_uri tests.test_backend_editor_contract 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_stylesheet_states tests.test_public_api -v
# Direct Qt/process integration and destruction/lifetime checks
python -m unittest tests.test_external_core tests.test_qt_lifetime -v
python -m unittest tests.test_application_process tests.test_external_core tests.test_layout_matrix tests.test_qt_lifetime -v
# Explicit slow stress tier
# Explicit repeated stress tier
python -m unittest tests.test_qt_stress tests.test_process_stress -v
# Explicit release-confidence tier (skipped unless opted in)
# PowerShell: $env:FURIOUS_VERY_HEAVY_TESTS = '1'
# POSIX shell: export FURIOUS_VERY_HEAVY_TESTS=1
python -m unittest tests.test_very_heavy -v
# Shared-state order-independence spot check
python -m unittest 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_backend_editor_contract tests.test_shadowsocks_uri tests.test_socks_uri tests.test_subscription_sync tests.test_subscription_manager tests.test_controllers tests.test_plugin_architecture tests.test_architecture_refactors tests.test_models_and_services tests.test_interface -v
python -m unittest tests.test_public_api tests.test_stylesheet_states 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_backend_editor_contract tests.test_shadowsocks_uri tests.test_socks_uri 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 discover -s tests -v
```
@@ -105,9 +111,11 @@ active Python interpreter. Two regressions cover Windows-specific behavior and
are skipped on other platforms; native handle/RSS trend checks use the safe
platform counters available on the current host.
Several negative-path tests intentionally emit warning or error log messages,
and the stress tier prints resource summaries. Treat the runner's final status
and exit code as authoritative; expected diagnostic output still ends in
Several negative-path tests intentionally emit warning or error log messages.
The opt-in release-confidence tier uses high operation counts but samples native
resources only at batch boundaries, so its assertions remain ownership-oriented.
Neither stress tier is enabled by a normal focused module run. Treat the runner's
final status and exit code as authoritative; expected diagnostic output still ends in
`OK`.
## Packaged-build smoke procedure
+57
View File
@@ -29,6 +29,9 @@ import uuid
import atexit
import ctypes
import tempfile
import sys
import subprocess
import threading
import weakref
# Select the headless platform before importing any Qt module. This process is
@@ -299,6 +302,60 @@ def currentNativeHandleCount() -> int | None:
return None
def childEnvironment(**overrides) -> dict[str, str]:
"""Return a hermetic child environment with offscreen Qt selected."""
environment = os.environ.copy()
environment['QT_QPA_PLATFORM'] = 'offscreen'
environment.update({key: str(value) for key, value in overrides.items()})
return environment
def runPythonChild(
script: str,
*,
timeout: float = 20,
environment: dict[str, str] | None = None,
) -> subprocess.CompletedProcess:
"""Run one exact child interpreter with bounded captured output."""
return subprocess.run(
[sys.executable, '-c', script],
capture_output=True,
env=environment or childEnvironment(),
text=True,
timeout=timeout,
)
def resourceSnapshot() -> dict[str, object]:
"""Return ownership-oriented process resource counters for stress batches."""
return {
'handles': currentNativeHandleCount(),
'rss': currentRSS(),
'threads': tuple(
sorted(
(thread.ident, thread.name)
for thread in threading.enumerate()
if thread.ident is not None
)
),
}
def veryHeavyEnabled() -> bool:
"""Return whether the explicit release-confidence stress tier is enabled."""
return os.environ.get('FURIOUS_VERY_HEAVY_TESTS') == '1'
def assertChildSucceeded(testCase, result, context: str = 'child process'):
"""Assert a child exited normally while preserving captured diagnostics."""
testCase.assertEqual(
result.returncode,
0,
f'{context}:\n{result.stdout}{result.stderr}',
)
@contextmanager
def isolatedSettings():
"""Route every QSettings read/write to one temporary test namespace."""
+261
View File
@@ -0,0 +1,261 @@
# Copyright (C) 2024–present Loren Eteval & contributors <loren.eteval@proton.me>
#
# This file is part of Furious.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Protect the outer application process and command-line dispatch boundaries."""
from __future__ import annotations
from Furious.Interface import ApplicationRunner
from Furious.Utility.AppMainProcess import AppMainProcess
from pathlib import Path
from types import SimpleNamespace
from unittest import mock
import importlib
import multiprocessing
import signal
import sys
import tempfile
import unittest
AppMainProcessModule = importlib.import_module('Furious.Utility.AppMainProcess')
class _SuccessfulApplication:
"""Provide one picklable process-boundary application fixture."""
@staticmethod
def run() -> int:
"""Return the normal semantic exit code."""
return ApplicationRunner.ExitCode.ExitSuccess.value
@staticmethod
def exit(_exitCode=0):
"""Accept the signal-handler contract without host side effects."""
def _successfulApplication():
"""Return one picklable application fixture in a spawned child."""
return _SuccessfulApplication()
class AppMainProcessTest(unittest.TestCase):
"""Verify exact child ownership, crash mapping, and signal safety."""
def testSharedCrashFlagDoesNotCreateAManagerServer(self):
"""Use one synchronized scalar without spawning an unmanaged manager."""
with mock.patch.object(
AppMainProcessModule.multiprocessing,
'Manager',
side_effect=AssertionError('manager must not be created'),
):
process = AppMainProcess(_successfulApplication)
try:
self.assertFalse(process.fileWritten.value)
finally:
process.close()
def testExactApplicationChildExitsWithoutAuxiliaryChildren(self):
"""Start, join, and close only the exact application child."""
baseline = {child.pid for child in multiprocessing.active_children()}
process = AppMainProcess(_successfulApplication)
try:
process.start()
process.join(15)
if process.is_alive():
process.terminate()
process.join(5)
self.fail('application child did not exit within its bound')
self.assertEqual(process.exitcode, 0)
self.assertFalse(process.fileWritten.value)
finally:
if process.is_alive():
process.terminate()
process.join(5)
process.close()
remaining = {child.pid for child in multiprocessing.active_children()}
self.assertEqual(remaining - baseline, set())
def testExceptionHookPreservesSemanticExitCodes(self):
"""Map assertion and unknown failures without hiding crash-log attempts."""
for exception, expected in (
(AssertionError('assertion'), ApplicationRunner.ExitCode.AssertionError),
(RuntimeError('runtime'), ApplicationRunner.ExitCode.UnknownException),
):
with self.subTest(exception=type(exception).__name__):
owner = SimpleNamespace(saveCrashLog=mock.Mock())
with (
mock.patch.object(AppMainProcessModule, 'APP', return_value=None),
mock.patch.object(
AppMainProcessModule.traceback,
'print_exception',
),
self.assertRaises(SystemExit) as raised,
):
AppMainProcess.exceptHook(
owner,
type(exception),
exception,
exception.__traceback__,
)
self.assertEqual(raised.exception.code, expected.value)
owner.saveCrashLog.assert_called_once()
def testExceptionHookUsesPartiallyConstructedApplicationWhenAvailable(self):
"""Request application-owned exit after construction has succeeded."""
application = SimpleNamespace(exit=mock.Mock())
owner = SimpleNamespace(saveCrashLog=mock.Mock())
exception = RuntimeError('runtime')
with (
mock.patch.object(
AppMainProcessModule,
'APP',
return_value=application,
),
mock.patch.object(AppMainProcessModule.traceback, 'print_exception'),
):
AppMainProcess.exceptHook(
owner,
type(exception),
exception,
exception.__traceback__,
)
application.exit.assert_called_once_with(
ApplicationRunner.ExitCode.UnknownException.value
)
def testSignalHandlerIsSafeBeforeAndAfterApplicationConstruction(self):
"""Handle the exact signal without dereferencing absent application state."""
owner = SimpleNamespace(application=None)
with self.assertRaises(SystemExit) as raised:
AppMainProcess.handler(owner, signal.SIGTERM, None)
self.assertEqual(
raised.exception.code,
ApplicationRunner.ExitCode.ExitSuccess.value,
)
application = SimpleNamespace(exit=mock.Mock())
owner.application = application
AppMainProcess.handler(owner, signal.SIGINT, None)
application.exit.assert_called_once_with()
def testCrashLogUsesOnlyTheConfiguredTemporaryPath(self):
"""Write one bounded diagnostic file and publish its shared result."""
owner = SimpleNamespace(
logFileName='crash.log',
fileWritten=SimpleNamespace(value=False),
)
try:
raise RuntimeError('diagnostic fixture')
except RuntimeError:
exceptionType, exceptionValue, tb = sys.exc_info()
with tempfile.TemporaryDirectory() as directory:
crashDirectory = Path(directory) / 'crashes'
with (
mock.patch.object(
AppMainProcessModule,
'CRASH_LOG_DIR',
crashDirectory,
),
mock.patch.object(AppMainProcessModule, 'APP', return_value=None),
):
AppMainProcess.saveCrashLog(
owner,
exceptionType,
exceptionValue,
tb,
)
self.assertTrue(owner.fileWritten.value)
self.assertIn(
'diagnostic fixture',
(crashDirectory / owner.logFileName).read_text(encoding='utf-8'),
)
class CommandLineDispatchTest(unittest.TestCase):
"""Keep entry-point selection narrow and diagnosable."""
def setUp(self):
"""Import the module after the test environment is established."""
self.module = importlib.import_module('Furious.__main__')
def testClearAndNormalCommandsUseOnlyTheirSelectedPath(self):
"""Dispatch the built-in clear command without starting the GUI child."""
with (
mock.patch.object(
self.module.sys,
'argv',
['Furious', self.module.AppBuiltinCommand.Clear.value],
),
mock.patch.object(self.module, 'runClearSettings') as clear,
mock.patch.object(self.module, 'runAppMain') as run,
):
self.module.main()
clear.assert_called_once_with()
run.assert_not_called()
with (
mock.patch.object(self.module.sys, 'argv', ['Furious']),
mock.patch.object(self.module, 'runClearSettings') as clear,
mock.patch.object(self.module, 'runAppMain') as run,
):
self.module.main()
run.assert_called_once_with()
clear.assert_not_called()
def testUnexpectedEntrypointFailureUsesStableFallbackExit(self):
"""Keep a bootstrap exception visible and map it to the legacy fallback."""
with (
mock.patch.object(self.module.sys, 'argv', ['Furious']),
mock.patch.object(
self.module,
'runAppMain',
side_effect=RuntimeError('bootstrap fixture'),
),
mock.patch.object(self.module.traceback, 'print_exc') as printException,
self.assertRaises(SystemExit) as raised,
):
self.module.main()
self.assertEqual(raised.exception.code, -1)
printException.assert_called_once_with()
if __name__ == '__main__':
unittest.main()
+203
View File
@@ -0,0 +1,203 @@
# Copyright (C) 2024–present Loren Eteval & contributors <loren.eteval@proton.me>
#
# This file is part of Furious.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Cover the complete Hysteria 1 protocol capability boundary."""
from __future__ import annotations
from Furious.Backends.Configuration import ConfigHysteria1
from Furious.Backends.Hysteria1.Plugin import (
Hysteria1CoreRuntimeFactory,
Hysteria1Plugin,
)
from Furious.Backends.Hysteria1.Protocols import HYSTERIA1_PROTOCOL_HANDLERS
from Furious.Plugins.API import CapabilityKind
from Furious.Plugins.Registry import PluginRegistry
from tests.support import application, collectAtBoundary, waitFor
from urllib.parse import quote
import copy
import unittest
import weakref
class Hysteria1ProtocolTest(unittest.TestCase):
"""Protect URI, mapping, editor, and runtime-copy compatibility."""
@classmethod
def setUpClass(cls):
"""Create Qt before asking the plugin for a transient editor."""
application()
def testCanonicalUriRoundTripPreservesAllSupportedFields(self):
"""Round-trip legacy flat fields, port hopping, and a Unicode remark."""
remark = 'Hysteria 1 + 测试'
uri = (
'hysteria://example.com:443'
'?mport=20000-30000'
'&protocol=faketcp'
'&auth=secret%3Avalue'
'&peer=sni.example'
'&insecure=1'
'&upmbps=55'
'&downmbps=120'
'&alpn=h3'
'&obfsParam=obfs%2Bvalue'
f'#{quote(remark, safe="")}'
)
handler = HYSTERIA1_PROTOCOL_HANDLERS[0]
parsed = handler.parse(uri)
self.assertIsNotNone(parsed)
self.assertEqual(parsed.metadata['displayName'], remark)
configuration = parsed.configuration
self.assertEqual(configuration['server'], 'example.com:443,20000-30000')
self.assertEqual(configuration['protocol'], 'faketcp')
self.assertEqual(configuration['auth_str'], 'secret:value')
self.assertEqual(configuration['server_name'], 'sni.example')
self.assertTrue(configuration['insecure'])
self.assertEqual(configuration['up_mbps'], 55)
self.assertEqual(configuration['down_mbps'], 120)
self.assertEqual(configuration['alpn'], 'h3')
self.assertEqual(configuration['obfs'], 'obfs+value')
reparsed = handler.parse(handler.export(configuration, remark))
self.assertIsNotNone(reparsed)
self.assertEqual(reparsed.metadata['displayName'], remark)
self.assertEqual(dict(reparsed.configuration), dict(configuration))
def testMalformedUrisFailWithoutPartialConfiguration(self):
"""Reject wrong schemes and non-numeric bandwidth through one codec."""
handler = HYSTERIA1_PROTOCOL_HANDLERS[0]
for uri in (
'hysteria2://example.com:443?upmbps=10&downmbps=20',
'hysteria://example.com:443?upmbps=bad&downmbps=20',
'hysteria://example.com:443?upmbps=10&downmbps=bad',
):
with self.subTest(uri=uri):
self.assertIsNone(handler.parse(uri))
def testMappingRecognitionAvoidsAmbiguousServerOnlyDocuments(self):
"""Recognize Hysteria 1 fields without claiming another core's mapping."""
handler = HYSTERIA1_PROTOCOL_HANDLERS[0]
self.assertIsNone(handler.fromMapping({'server': 'example.com:443'}))
configuration = handler.fromMapping(
{
'server': 'example.com:443',
'protocol': 'udp',
'futureField': {'preserved': True},
}
)
self.assertIsInstance(configuration, ConfigHysteria1)
self.assertEqual(
configuration['futureField'],
{'preserved': True},
)
def testDownloadPreparationUsesAnIndependentFullDocument(self):
"""Strip only SOCKS from the derived probe configuration."""
original = ConfigHysteria1(
{
'server': 'example.com:443',
'protocol': 'udp',
'up_mbps': 20,
'down_mbps': 80,
'http': {'listen': '127.0.0.1:10809'},
'socks5': {'listen': '127.0.0.1:10808'},
'futureField': {'preserved': True},
}
)
before = copy.deepcopy(dict(original))
prepared = Hysteria1CoreRuntimeFactory().prepareDownloadTest(original, 19090)
self.assertEqual(dict(original), before)
self.assertIsNot(prepared, original)
self.assertEqual(prepared['http']['listen'], '127.0.0.1:19090')
self.assertNotIn('socks5', prepared)
self.assertEqual(prepared['futureField'], {'preserved': True})
def testPluginDiscoversProtocolEditorAndRuntimeCapabilities(self):
"""Keep Hysteria 1 entirely behind the generic plugin registry."""
registry = PluginRegistry()
plugin = Hysteria1Plugin()
registry.register(plugin)
try:
self.assertIs(
registry.plugin('official.hysteria1'),
plugin,
)
self.assertEqual(
len(
registry.capabilities(
CapabilityKind.Protocol,
plugin=plugin,
)
),
1,
)
self.assertEqual(
len(
registry.capabilities(
CapabilityKind.ProtocolEditor,
plugin=plugin,
)
),
1,
)
self.assertEqual(
len(
registry.capabilities(
CapabilityKind.CoreRuntimeFactory,
plugin=plugin,
)
),
1,
)
configuration = ConfigHysteria1(
{
'server': 'example.com:443',
'protocol': 'udp',
'up_mbps': 20,
'down_mbps': 80,
}
)
editor = registry.createEditorForConfig(configuration)
reference = weakref.ref(editor)
editor.show()
editor.close()
editor = None
self.assertTrue(waitFor(lambda: reference() is None))
finally:
registry.shutdown()
collectAtBoundary()
if __name__ == '__main__':
unittest.main()
+115
View File
@@ -0,0 +1,115 @@
# Copyright (C) 2024–present Loren Eteval & contributors <loren.eteval@proton.me>
#
# This file is part of Furious.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Exercise representative Fluent layouts in isolated display-scale processes."""
from __future__ import annotations
from tests.support import assertChildSucceeded, childEnvironment, runPythonChild
import unittest
class IsolatedDisplayMatrixTest(unittest.TestCase):
"""Keep navigation and transient dialogs stable across supported scales."""
Script = r"""
from Furious.Qt import AppQMessageBox, AppStyleSheet
from Furious.Widget import NavigationView
from PySide6 import QtCore
from PySide6.QtWidgets import QWidget
from tests.support import application, collectAtBoundary, processQtEvents
app = application()
navigation = NavigationView()
navigation.resize(960, 640)
for index in range(5):
navigation.addPage(
f'page-{index}',
QWidget(parent=navigation),
f'Page {index} with a translated title',
'house-door.svg',
)
navigation.show()
processQtEvents()
for theme in (AppStyleSheet.Light, AppStyleSheet.Dark):
app.setStyleSheet(AppStyleSheet.forTheme(theme))
for expanded in (True, False, True):
navigation.setExpanded(expanded, animated=False)
processQtEvents()
assert navigation.width() > 0
assert navigation.height() > 0
for index in range(5):
page_id = f'page-{index}'
navigation.setCurrentPage(page_id)
processQtEvents()
assert navigation.currentPageId() == page_id
assert navigation.pageStack.currentWidget() is navigation.page(page_id)
message_box = AppQMessageBox(
icon=AppQMessageBox.Icon.Information,
parent=navigation,
heading='Scale and theme verification',
text='A sufficiently long localized message must remain readable.',
buttons=(
AppQMessageBox.StandardButton.Ok
| AppQMessageBox.StandardButton.Cancel
),
)
message_box.show()
processQtEvents()
assert message_box.width() >= 300
assert message_box.height() >= 120
assert message_box.surface.geometry().contains(
message_box.contentFrame.geometry().center()
)
message_box.close()
message_box.deleteLater()
collectAtBoundary()
navigation.close()
navigation.deleteLater()
collectAtBoundary()
assert not AppQMessageBox._openMessageBoxes
"""
def testScaleAndThemeMatrix(self):
"""Run each scale before QApplication exists to honor Qt semantics."""
for scale in ('1', '1.25', '1.5', '2'):
with self.subTest(scale=scale):
result = runPythonChild(
self.Script,
environment=childEnvironment(QT_SCALE_FACTOR=scale),
)
assertChildSucceeded(self, result, f'display scale {scale}')
if __name__ == '__main__':
unittest.main()
+23
View File
@@ -278,6 +278,29 @@ class MainWindowGeometryTest(unittest.TestCase):
self.assertEqual(window.pos(), movedPosition)
moveToCenter.assert_called_once_with(window)
expectedPages = {
'home': window.homePage,
'log': window.logPage,
'subscription': window.subscriptionPage,
'metrics': window.metricsPage,
'settings': window.settingsPage,
}
for pageId, page in expectedPages.items():
with self.subTest(pageId=pageId):
window.showPage(pageId)
processQtEvents()
self.assertEqual(
window.navigationView.currentPageId(),
pageId,
)
self.assertIs(window.navigationView.page(pageId), page)
self.assertIs(
window.navigationView.pageStack.currentWidget(),
page,
)
window.close()
window.deleteLater()
window = None
+164
View File
@@ -0,0 +1,164 @@
# Copyright (C) 2024–present Loren Eteval & contributors <loren.eteval@proton.me>
#
# This file is part of Furious.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Protect curated package exports and import-time isolation."""
from __future__ import annotations
from tests.support import assertChildSucceeded, runPythonChild
import unittest
class PublicPackageAPITest(unittest.TestCase):
"""Keep the documented package-level import surface deliberate."""
Packages = (
'Actions',
'Application',
'Backends',
'Controllers',
'Core',
'Extensions',
'Externals',
'Frozenlib',
'Interface',
'Models',
'Plugins',
'Qt',
'Repository',
'Service',
'Utility',
'Widget',
'Window',
)
def testEveryPackageDeclaresAndResolvesItsPublicAPI(self):
"""Make every curated star export importable in one isolated process."""
script = f"""
import importlib
packages = {self.Packages!r}
for package_name in packages:
module_name = f'Furious.{{package_name}}'
module = importlib.import_module(module_name)
exports = getattr(module, '__all__', None)
assert isinstance(exports, list), (module_name, exports)
assert exports, module_name
assert len(exports) == len(set(exports)), module_name
namespace = {{}}
exec(f'from {{module_name}} import *', namespace)
actual = set(namespace).difference({{'__builtins__'}})
assert actual == set(exports), (module_name, sorted(actual), sorted(exports))
"""
result = runPythonChild(script)
assertChildSucceeded(self, result, 'public package import child')
def testRepresentativeCompatibilitySymbolsRemainStable(self):
"""Pin the architectural entry points used across package boundaries."""
script = """
from Furious.Application import DesktopApplication, TrayIcon
from Furious.Controllers import (
ConnectionController,
RoutingController,
SettingsController,
)
from Furious.Interface import (
ApplicationRunner,
CoreRuntime,
EditorBinding,
StorageBackend,
)
from Furious.Models import CoreConfiguration, ProfileMetadata, ServerProfile
from Furious.Plugins import CoreRuntimeFactory, FuriousPlugin, PluginRegistry
from Furious.Qt import AppQDialog, AppQMainWindow, AppQMessageBox
from Furious.Repository import Storage, SubscriptionGroup
from Furious.Service import (
ConnectionManager,
LogManager,
MetricsHistory,
SubscriptionManager,
)
from Furious.Utility import AppMainProcess
from Furious.Widget import NavigationView
from Furious.Window import MainWindow, SettingsPage
symbols = (
DesktopApplication,
TrayIcon,
ConnectionController,
RoutingController,
SettingsController,
ApplicationRunner,
CoreRuntime,
EditorBinding,
StorageBackend,
CoreConfiguration,
ProfileMetadata,
ServerProfile,
CoreRuntimeFactory,
FuriousPlugin,
PluginRegistry,
AppQDialog,
AppQMainWindow,
AppQMessageBox,
Storage,
SubscriptionGroup,
ConnectionManager,
LogManager,
MetricsHistory,
SubscriptionManager,
AppMainProcess,
NavigationView,
MainWindow,
SettingsPage,
)
assert all(symbol is not None for symbol in symbols)
"""
result = runPythonChild(script)
assertChildSucceeded(self, result, 'representative public API child')
def testBackendPluginDiscoveryStaysLazyAndComplete(self):
"""Resolve bundled plugin types without constructing plugin instances."""
script = """
import sys
import Furious.Backends as backends
assert 'Furious.Backends.Xray.Plugin' not in sys.modules
plugin_types = backends.OFFICIAL_PLUGIN_TYPES
assert tuple(plugin.__name__ for plugin in plugin_types) == (
'XrayPlugin',
'Hysteria1Plugin',
'Hysteria2Plugin',
'ExternalCorePlugin',
)
"""
result = runPythonChild(script)
assertChildSucceeded(self, result, 'lazy backend discovery child')
if __name__ == '__main__':
unittest.main()
+187
View File
@@ -0,0 +1,187 @@
# Copyright (C) 2024–present Loren Eteval & contributors <loren.eteval@proton.me>
#
# This file is part of Furious.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Protect repository isolation, live-collection, and compatibility contracts."""
from __future__ import annotations
from Furious.Frozenlib import AppSettings
from Furious.Models import CoreConfiguration, ServerProfile
from Furious.Models.Encoding import PyBase64Encoder, UJSONEncoder
from Furious.Repository.Routings import UserRoutings
from Furious.Repository.Storage import Storage
from Furious.Repository.TunSettings import UserTUNSettings
from tests.support import application, isolatedSettings
from types import SimpleNamespace
from unittest import mock
import importlib
import unittest
StorageModule = importlib.import_module('Furious.Repository.Storage')
class RepositoryContractTest(unittest.TestCase):
"""Verify every persisted collection stays isolated and backward compatible."""
@classmethod
def setUpClass(cls):
"""Create the application required by the process-lifetime storage cache."""
application()
def tearDown(self):
"""Release application-lifetime repository cache entries between tests."""
self._clearStorageCaches()
@staticmethod
def _clearStorageCaches():
"""Clear only finite cache entries created by this test process."""
for name in (
'_UserServersStorage',
'_UserSubsStorage',
'_UserTUNSettingsStorage',
'_UserRoutingsStorage',
):
getattr(Storage, name).cache_clear()
def testRoutingRepositoryRoundTripPreservesUnknownDocuments(self):
"""Persist arbitrary core-owned routing fields without normalization."""
with isolatedSettings():
repository = UserRoutings()
repository.data()['custom-id'] = {
'remark': 'Custom route',
'routing': {
'domainStrategy': 'future-strategy',
'futureField': {'enabled': True},
},
}
repository.sync()
restored = UserRoutings().data()
self.assertEqual(restored, repository.data())
self.assertEqual(
restored['custom-id']['routing']['futureField'],
{'enabled': True},
)
def testTunSettingsRoundTripAndRejectsNonObjectRoots(self):
"""Keep valid unknown fields while treating incompatible roots as empty."""
with isolatedSettings():
repository = UserTUNSettings()
repository.data().update(
{
'interfaceName': 'fixture-tun',
'futureOption': {'mode': 'future'},
}
)
repository.sync()
self.assertEqual(UserTUNSettings().data(), repository.data())
encodedList = PyBase64Encoder.encode(
UJSONEncoder.encode(['not', 'an', 'object']).encode()
)
AppSettings.set('CustomTUNSettings', encodedList)
self.assertEqual(UserTUNSettings().data(), {})
def testStorageCachesOneLiveOwnerPerRepositoryType(self):
"""Return one intentional process-lifetime owner and its live collection."""
with isolatedSettings():
self._clearStorageCaches()
routings = Storage.UserRoutings()
tunSettings = Storage.UserTUNSettings()
self.assertIs(routings, Storage.UserRoutings())
self.assertIs(tunSettings, Storage.UserTUNSettings())
routings['fixture'] = {'remark': 'Route'}
tunSettings['interfaceName'] = 'fixture-tun'
self.assertEqual(Storage.UserRoutings()['fixture']['remark'], 'Route')
self.assertEqual(
Storage.UserTUNSettings()['interfaceName'],
'fixture-tun',
)
def testActiveServerRemarkUsesTheExactControllerProfile(self):
"""Render the selected live profile instead of swallowing a name error."""
class DisplayConfiguration(CoreConfiguration):
@property
def itemRemark(self):
"""Expose the profile remark through the configuration contract."""
return 'Active profile'
profile = ServerProfile.fromConfiguration(
DisplayConfiguration({'type': 'fixture'}),
{'displayName': 'Active profile'},
)
controller = SimpleNamespace(
activeProfile=profile,
isConnected=mock.Mock(return_value=True),
)
with (
mock.patch.object(
StorageModule,
'AppConnectionController',
return_value=controller,
),
mock.patch.object(Storage, 'UserServers', return_value=[profile]),
):
self.assertEqual(
Storage.Extras.UserServerRemark(),
'1 - Active profile',
)
def testActiveServerRemarkOmitsIndexForNonRepositoryProfile(self):
"""Keep an active ad-hoc profile visible without inventing a row index."""
profile = ServerProfile.fromConfiguration(
CoreConfiguration({'type': 'fixture'}),
{'displayName': 'Ad hoc'},
)
controller = SimpleNamespace(
activeProfile=profile,
isConnected=mock.Mock(return_value=True),
)
with (
mock.patch.object(
StorageModule,
'AppConnectionController',
return_value=controller,
),
mock.patch.object(Storage, 'UserServers', return_value=[]),
):
self.assertEqual(Storage.Extras.UserServerRemark(), 'Ad hoc')
def testActivatedIndexMalformedInputFallsBackWithoutMutation(self):
"""Treat incompatible persisted input as no selection."""
with isolatedSettings() as settings:
settings.setValue('ActivatedItemIndex', 'not-an-index')
self.assertEqual(Storage.UserActivatedItemIndex(), -1)
self.assertEqual(settings.value('ActivatedItemIndex'), 'not-an-index')
if __name__ == '__main__':
unittest.main()
+357
View File
@@ -0,0 +1,357 @@
# Copyright (C) 2024–present Loren Eteval & contributors <loren.eteval@proton.me>
#
# This file is part of Furious.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Provide an explicit release-confidence tier for sustained churn."""
from __future__ import annotations
from Furious.Backends.ExternalCore import ConfigExternalCore, ExternalCoreProcess
from Furious.Interface import ApplicationRunner
from Furious.Models import LogCategory
from Furious.Plugins import FuriousPlugin, PluginMetadata, PluginRegistry
from Furious.Qt import AppQMessageBox, AppQTransientDialog
from Furious.Utility import AppMainProcess
from Furious.Service import APPLICATION_LOG_CATEGORY, LogManager, MetricsHistory
from Furious.Widget import NavigationView
from PySide6.QtWidgets import QWidget
from tests.support import (
application,
collectAtBoundary,
processQtEvents,
resourceSnapshot,
veryHeavyEnabled,
)
from pathlib import Path
import gc
import multiprocessing
import sys
import tempfile
import threading
import unittest
import weakref
class _ReleaseApplication:
"""Provide one picklable successful application-process fixture."""
@staticmethod
def run() -> int:
"""Return the normal semantic application exit code."""
return ApplicationRunner.ExitCode.ExitSuccess.value
def _releaseApplication():
"""Return one application fixture inside the spawned child."""
return _ReleaseApplication()
@unittest.skipUnless(
veryHeavyEnabled(),
'set FURIOUS_VERY_HEAVY_TESTS=1 for release-confidence stress tests',
)
class VeryHeavyContractTest(unittest.TestCase):
"""Exercise high-count pure and Qt ownership contracts on explicit request."""
@classmethod
def setUpClass(cls):
"""Create the one process-lifetime test application."""
application()
def tearDown(self):
"""Drain deferred deletion before the next sustained workload."""
collectAtBoundary()
def testFiveThousandPluginLifecyclesReleaseEachRegistry(self):
"""Register, shut down, and collect independent plugin owners."""
stopped = 0
pluginReferences = []
registryReferences = []
snapshots = []
class FixturePlugin(FuriousPlugin):
"""Expose metadata only and record exact lifecycle completion."""
def __init__(self, index):
"""Give every plugin one non-conflicting stable identity."""
self.metadata = PluginMetadata(f'tests.heavy.{index}', 'Heavy')
def shutdown(self):
"""Record the registry-owned shutdown callback."""
nonlocal stopped
stopped += 1
for index in range(5_000):
registry = PluginRegistry()
plugin = FixturePlugin(index)
registry.register(plugin)
registry.shutdown()
pluginReferences.append(weakref.ref(plugin))
registryReferences.append(weakref.ref(registry))
del plugin, registry
if (index + 1) % 1_000 == 0:
gc.collect()
snapshots.append(resourceSnapshot())
gc.collect()
self.assertEqual(stopped, 5_000)
self.assertTrue(all(reference() is None for reference in pluginReferences))
self.assertTrue(all(reference() is None for reference in registryReferences))
self.assertEqual(snapshots[-1]['threads'], snapshots[0]['threads'])
def testOneHundredApplicationChildrenLeaveNoAuxiliaryProcesses(self):
"""Repeat the exact outer child boundary without manager processes."""
baselineChildren = {child.pid for child in multiprocessing.active_children()}
baseline = resourceSnapshot()
snapshots = []
for index in range(100):
process = AppMainProcess(_releaseApplication)
try:
process.start()
process.join(15)
if process.is_alive():
process.terminate()
process.join(5)
self.fail(f'application child {index} exceeded its timeout')
self.assertEqual(process.exitcode, 0)
self.assertFalse(process.fileWritten.value)
finally:
if process.is_alive():
process.terminate()
process.join(5)
process.close()
if (index + 1) % 20 == 0:
snapshots.append(resourceSnapshot())
remainingChildren = {
child.pid for child in multiprocessing.active_children()
}.difference(baselineChildren)
self.assertEqual(remainingChildren, set())
self.assertEqual(snapshots[-1]['threads'], baseline['threads'])
def testOneHundredExternalCoreCyclesReapPipesAndThreads(self):
"""Sustain the real harmless subprocess lifecycle at release scale."""
baselineThreads = {thread.ident for thread in threading.enumerate()}
baseline = resourceSnapshot()
snapshots = []
with tempfile.TemporaryDirectory(dir=Path.cwd()) as directory:
configuration = ConfigExternalCore(
{
'type': 'external-core',
'executable': str(Path(sys.executable).resolve()),
'workingDirectory': directory,
'arguments': ['-u', '-c', 'import time; time.sleep(60)'],
'environment': {},
'httpProxy': '127.0.0.1:10809',
'socksProxy': '127.0.0.1:10808',
'shutdownTimeout': 1,
}
)
runtime = ExternalCoreProcess()
try:
for index in range(100):
self.assertTrue(runtime.start(configuration))
process = runtime.process
self.assertIsNotNone(process)
self.assertTrue(runtime.isAlive())
runtime.stop()
self.assertFalse(runtime.isAlive())
self.assertIsNone(runtime.process)
self.assertFalse(runtime._readerThreads)
self.assertIsNone(runtime._watcherThread)
self.assertIsNotNone(process.poll())
if (index + 1) % 20 == 0:
snapshots.append(resourceSnapshot())
finally:
runtime.dispose()
remainingThreads = {
thread.ident
for thread in threading.enumerate()
if thread.ident not in baselineThreads
}
self.assertEqual(remainingThreads, set())
self.assertEqual(snapshots[-1]['threads'], baseline['threads'])
baselineHandles = baseline['handles']
handles = tuple(
snapshot['handles']
for snapshot in snapshots
if snapshot['handles'] is not None
)
if baselineHandles is not None and len(handles) == len(snapshots):
self.assertLessEqual(
max((*handles, baselineHandles)) - baselineHandles,
4,
)
def testOneHundredThousandMetricsRemainTimeBounded(self):
"""Prune a long monotonic stream to the configured history window."""
history = MetricsHistory(maximumHistorySeconds=1_000)
for sampledAt in range(100_000):
history.recordSample(
{'network.download.speed': sampledAt},
sampledAt=sampledAt,
)
samples = history.rawSamples()
self.assertEqual(len(samples), 1_001)
self.assertEqual(samples[0].sampledAt, 98_999)
self.assertEqual(samples[-1].sampledAt, 99_999)
history.deleteLater()
def testConcurrentFortyThousandLogsHaveOneTotalOrder(self):
"""Preserve every entry exactly once under concurrent producers."""
manager = LogManager(maximumEntries=40_000, autoClearEnabled=False)
manager.registerCategory(LogCategory('tests.concurrent', 'Concurrent'))
baselineThreads = resourceSnapshot()['threads']
workers = []
def appendBatch(workerIndex):
"""Append one disjoint producer range through the public API."""
for entryIndex in range(5_000):
manager.append(
f'{workerIndex}:{entryIndex}',
'tests.concurrent',
)
for workerIndex in range(8):
worker = threading.Thread(target=appendBatch, args=(workerIndex,))
worker.start()
workers.append(worker)
for worker in workers:
worker.join(30)
self.assertFalse(worker.is_alive())
entries = manager.entries('tests.concurrent')
self.assertEqual(len(entries), 40_000)
self.assertEqual(len({entry.message for entry in entries}), 40_000)
self.assertEqual(
tuple(entry.sequence for entry in entries),
tuple(range(1, 40_001)),
)
self.assertEqual(resourceSnapshot()['threads'], baselineThreads)
manager.deleteLater()
def testTwentyThousandNavigationSwitchesPreserveOnePage(self):
"""Switch existing pages without allocating replacement widgets."""
navigation = NavigationView()
pages = []
for index in range(16):
page = QWidget(parent=navigation)
pages.append(page)
navigation.addPage(
f'page-{index}',
page,
f'Page {index}',
'house-door.svg',
)
navigation.show()
processQtEvents()
for index in range(20_000):
navigation.setCurrentPage(f'page-{index % len(pages)}')
if (index + 1) % 2_000 == 0:
processQtEvents()
self.assertEqual(navigation.currentPageId(), 'page-15')
self.assertIs(navigation.pageStack.currentWidget(), pages[-1])
self.assertEqual(navigation.pageStack.count(), len(pages))
navigation.close()
navigation.deleteLater()
def testOneThousandTransientDialogsLeaveNoRegistryEntries(self):
"""Open and destroy dialogs in batches without retaining wrappers."""
references = []
baseline = resourceSnapshot()
for index in range(1_000):
dialog = AppQTransientDialog()
references.append(weakref.ref(dialog))
dialog.show()
dialog.close()
dialog.deleteLater()
if (index + 1) % 50 == 0:
collectAtBoundary()
del dialog
collectAtBoundary()
self.assertTrue(all(reference() is None for reference in references))
self.assertFalse(AppQTransientDialog._openDialogs)
self.assertEqual(resourceSnapshot()['threads'], baseline['threads'])
def testFiveHundredMessageBoxesLeaveNoRegistryEntries(self):
"""Exercise the specialized asynchronous registry at release scale."""
references = []
for index in range(500):
messageBox = AppQMessageBox(
icon=AppQMessageBox.Icon.Information,
text=f'Message {index}',
buttons=AppQMessageBox.StandardButton.Ok,
)
references.append(weakref.ref(messageBox))
messageBox.show()
messageBox.accept()
messageBox.deleteLater()
if (index + 1) % 25 == 0:
collectAtBoundary()
del messageBox
collectAtBoundary()
self.assertTrue(all(reference() is None for reference in references))
self.assertFalse(AppQMessageBox._openMessageBoxes)
if __name__ == '__main__':
unittest.main()