diff --git a/Furious/Backends/Configuration.py b/Furious/Backends/Configuration.py index bed62639..6ecd69a2 100644 --- a/Furious/Backends/Configuration.py +++ b/Furious/Backends/Configuration.py @@ -34,6 +34,7 @@ from Furious.Backends.SocksURI import ( parseSocksURI, serializeSocksURI, ) +from Furious.Frozenlib.PythonCompatibility import PythonCompatibility from Furious.Frozenlib.SocksProxy import socksURL from typing import Union, Tuple @@ -1614,7 +1615,7 @@ class ConfigXray(CoreConfiguration): + ':' + quote(password, safe='') + '@' - + endpoint.removeprefix('socks5://') + + PythonCompatibility.removePrefix(endpoint, 'socks5://') ) raise ValueError('A SOCKS listener is required for application TUN') @@ -1992,7 +1993,7 @@ class ConfigHysteria1(CoreConfiguration): + ':' + quote(password, safe='') + '@' - + endpoint.removeprefix('socks5://') + + PythonCompatibility.removePrefix(endpoint, 'socks5://') ) def setHttpProxy(self, endpoint: str) -> bool: @@ -2378,7 +2379,7 @@ class ConfigHysteria2(CoreConfiguration): + ':' + quote(password, safe='') + '@' - + endpoint.removeprefix('socks5://') + + PythonCompatibility.removePrefix(endpoint, 'socks5://') ) def setHttpProxy(self, endpoint: str) -> bool: diff --git a/Furious/Core/SingTUN.py b/Furious/Core/SingTUN.py index 4e987e92..0a18d408 100644 --- a/Furious/Core/SingTUN.py +++ b/Furious/Core/SingTUN.py @@ -24,10 +24,10 @@ from Furious.Core.MultiprocessingRuntime import ( ProcessLaunchSpec, ) from Furious.Core.ProcessOutput import ProcessOutputRedirector +from Furious.Frozenlib.PythonCompatibility import PythonCompatibility from Furious.Interface import RuntimeExit, RuntimeExitReason, RuntimeStartError import copy -import importlib.metadata import json import multiprocessing import threading @@ -197,8 +197,8 @@ class SingTUN(MultiprocessingRuntime): @staticmethod def version(): try: - return importlib.metadata.version('sing-tun') - except importlib.metadata.PackageNotFoundError: + return PythonCompatibility.distributionVersion('sing-tun') + except PythonCompatibility.PackageNotFoundError: return 'unavailable' def start(self): diff --git a/Furious/Frozenlib/PythonCompatibility.py b/Furious/Frozenlib/PythonCompatibility.py new file mode 100644 index 00000000..0c76a305 --- /dev/null +++ b/Furious/Frozenlib/PythonCompatibility.py @@ -0,0 +1,115 @@ +# Copyright (C) 2024–present Loren Eteval & contributors +# +# This file is part of Furious. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +"""Provide standard-library operations shared across supported Python versions.""" + +from __future__ import annotations + +from concurrent.futures import Executor, Future +from importlib import metadata +from typing import Iterable + +import sys + +__all__ = ['PythonCompatibility'] + +_PYTHON_39_OR_NEWER = sys.version_info >= (3, 9) + + +if _PYTHON_39_OR_NEWER: + _removePrefix = str.removeprefix + _removeSuffix = str.removesuffix + + def _shutdownExecutor(executor, pendingFutures, *, wait): + """Use native queued-work cancellation on Python 3.9 and newer.""" + executor.shutdown(wait=wait, cancel_futures=True) + +else: + + def _removePrefix(value: str, prefix: str) -> str: + """Remove one exact prefix on Python 3.8.""" + if not isinstance(prefix, str): + raise TypeError('prefix must be a string') + + return value[len(prefix) :] if value.startswith(prefix) else value + + def _removeSuffix(value: str, suffix: str) -> str: + """Remove one exact nonempty suffix on Python 3.8.""" + if not isinstance(suffix, str): + raise TypeError('suffix must be a string') + + return value[: -len(suffix)] if suffix and value.endswith(suffix) else value + + def _shutdownExecutor(executor, pendingFutures, *, wait): + """Cancel caller-owned queued futures before Python 3.8 shutdown.""" + for future in pendingFutures: + future.cancel() + + executor.shutdown(wait=wait) + + +if hasattr(metadata, 'EntryPoints'): + + def _entryPoints(group: str) -> tuple: + """Use selectable entry points without an import-time metadata scan.""" + return tuple(metadata.entry_points(group=group)) + +else: + + def _entryPoints(group: str) -> tuple: + """Select from the legacy metadata mapping without an import-time scan.""" + return tuple(metadata.entry_points().get(group, ())) + + +class PythonCompatibility: + """Keep version-dependent standard-library behavior at one boundary.""" + + PackageNotFoundError = metadata.PackageNotFoundError + + @staticmethod + def removePrefix(value: str, prefix: str) -> str: + """Remove one exact prefix using the implementation selected at import.""" + return _removePrefix(value, prefix) + + @staticmethod + def removeSuffix(value: str, suffix: str) -> str: + """Remove one exact suffix using the implementation selected at import.""" + return _removeSuffix(value, suffix) + + @staticmethod + def distributionVersion(name: str) -> str: + """Read installed metadata, preserving missing-distribution diagnostics.""" + return metadata.version(name) + + @staticmethod + def entryPoints(group: str) -> tuple: + """Fetch a group using the metadata API selected at import.""" + return _entryPoints(group) + + @staticmethod + def shutdownExecutor( + executor: Executor, + pendingFutures: Iterable[Future], + *, + wait: bool = True, + ) -> None: + """Cancel queued work and close admission without terminating running calls. + + The caller closes its own submission path before invoking this method and + supplies all uncancelled pending futures for the Python 3.8 fallback. + """ + _shutdownExecutor(executor, pendingFutures, wait=wait) diff --git a/Furious/Frozenlib/__init__.py b/Furious/Frozenlib/__init__.py index 61ed1ce6..b496852e 100644 --- a/Furious/Frozenlib/__init__.py +++ b/Furious/Frozenlib/__init__.py @@ -89,6 +89,7 @@ from .Globals import ( AppTrafficStatsManager, ) from .Mixins import Mixins +from .PythonCompatibility import PythonCompatibility from .PySide6Legacy import PySide6Legacy from .StartupOnBoot import StartupOnBoot from .SystemProxy import SystemProxy @@ -164,6 +165,7 @@ __all__ = [ 'PLATFORM_RELEASE', 'PROXY_SERVER_BYPASS', 'PYSIDE6_VERSION', + 'PythonCompatibility', 'PySide6Legacy', 'ROOT_DIR', 'SYSTEM_LANGUAGE', diff --git a/Furious/Models/Profile.py b/Furious/Models/Profile.py index 9ec73e19..a335bdc6 100644 --- a/Furious/Models/Profile.py +++ b/Furious/Models/Profile.py @@ -19,9 +19,9 @@ from __future__ import annotations -from collections.abc import Iterator, Mapping, MutableMapping +from collections.abc import Iterator, Mapping from dataclasses import dataclass, field -from typing import Any +from typing import Any, MutableMapping import copy import hashlib diff --git a/Furious/Plugins/Registry.py b/Furious/Plugins/Registry.py index 34eeb0a0..4f92990d 100644 --- a/Furious/Plugins/Registry.py +++ b/Furious/Plugins/Registry.py @@ -19,9 +19,10 @@ from __future__ import annotations +from Furious.Frozenlib.PythonCompatibility import PythonCompatibility + from collections.abc import Mapping from dataclasses import dataclass, field -from importlib import metadata from typing import Optional from urllib.parse import urlsplit @@ -1293,12 +1294,7 @@ class PluginRegistry: def discover(self): """Load trusted third-party plugins exposed through entry points.""" try: - entryPoints = metadata.entry_points() - - if hasattr(entryPoints, 'select'): - entryPoints = entryPoints.select(group=PLUGIN_ENTRY_POINT_GROUP) - else: - entryPoints = entryPoints.get(PLUGIN_ENTRY_POINT_GROUP, tuple()) + entryPoints = PythonCompatibility.entryPoints(PLUGIN_ENTRY_POINT_GROUP) except Exception as ex: # Any non-exit exceptions diff --git a/Furious/Service/TrafficStatsManager.py b/Furious/Service/TrafficStatsManager.py index 81c1e702..5b4a51bd 100644 --- a/Furious/Service/TrafficStatsManager.py +++ b/Furious/Service/TrafficStatsManager.py @@ -25,6 +25,7 @@ from Furious.Frozenlib import ( AppBinarySettings, AppSettings, Mixins, + PythonCompatibility, registerAppSettings, ) from Furious.Plugins import TrafficCounters, getPluginRegistry @@ -280,12 +281,17 @@ class TrafficStatsManager( def _closeExecutor(self): """Cancel queued work and release the background query executor.""" executor = self._executor + future = self._future self._cancelCurrentQuery() self._executor = None if executor is not None: - executor.shutdown(wait=False, cancel_futures=True) + PythonCompatibility.shutdownExecutor( + executor, + (future,) if future is not None else (), + wait=False, + ) def _cancelCurrentQuery(self): """Forget the active generation and cancel it when still queued.""" diff --git a/Translation.py b/Translation.py index cd6d02e1..06830ed4 100644 --- a/Translation.py +++ b/Translation.py @@ -57,7 +57,9 @@ def getAppSourceCodePath(path): @functools.lru_cache(None) def getMagicNameFromPath(path): - return os.path.relpath(path, ROOT_DIR).removesuffix('.py').replace(os.sep, '.') + return PythonCompatibility.removeSuffix( + os.path.relpath(path, ROOT_DIR), '.py' + ).replace(os.sep, '.') @functools.lru_cache(None) diff --git a/tests/README.md b/tests/README.md index fa1634ff..a1bc7b09 100644 --- a/tests/README.md +++ b/tests/README.md @@ -92,6 +92,7 @@ worker. Choose tests by the changed contract rather than by filename alone. | [test_controllers.py](test_controllers.py) | Connection state/error/reconnect transitions, reentrant cancellation/replacement before launch and during completion, startup restoration, shared settings, routing fallback persistence and tray/selector agreement after custom-routing disable/re-enable. | | [test_runtime_lifecycle.py](test_runtime_lifecycle.py) | Qt-thread exit dispatch, commit/exit races, duplicate and late exits, idempotent release and retained failure/retry ownership, failed reap/handle close, spawn failure, queue/timer disposal on preparation failure. | | [test_external_core.py](test_external_core.py) | Harmless real process launch/output/shutdown, partial thread-start rollback, non-finite timeout rejection, failed reap/thread-join retry, readiness/TUN metadata, Windows paths with spaces, subscription rejection of executable profiles, bounded DNS references. | +| [test_python_compatibility.py](test_python_compatibility.py) | Once-only implementation selection without import-time metadata enumeration, exact affix removal, both metadata entry-point APIs and plugin discovery, distribution-version failures, real modern/legacy executor cancellation and statistics-owner shutdown, and profile import with an unsubscriptable collections ABC. Legacy API shapes are simulated; this does not replace execution on the minimum interpreter. | | [test_frozenlib.py](test_frozenlib.py) | Nested state guards, cleanup isolation, native peer destruction during connection/theme/translation/cleanup notifications, bounded caches/throttling, dual-stack probe selection, mocked proxy/DNS/routes/startup/session boundaries and failure handling. | | [test_native_tun_semantics.py](test_native_tun_semantics.py) | Xray/Hysteria2 runtime-copy TUN preservation/replacement, managed-TUN failures, download-test stripping, prevention of a second tun2socks owner. | | [test_sing_tun.py](test_sing_tun.py) | tun2socks engine default, saved engine choices and ownership-aware reconnect notices, sing-tun Go stack default, development-stack and congestion-control compatibility, independent backend settings and host defaults, selected-only snapshots, captured TUN log sources in both startup paths, staged settings and malformed storage, SOCKS authentication/UDP, harmless real spawn/status/stop, native readiness, cancellation, synchronous compatibility, mocked DNS/routing ownership including development IPv6 rule ordering and refused cleanup. | @@ -138,7 +139,7 @@ regular cases run by default while its heavy class remains gated. ```text # Models, storage, plugin API, protocol/editor compatibility -python -m unittest tests.test_interface tests.test_models_and_services tests.test_repository_contracts tests.test_plugin_architecture tests.test_public_api tests.test_hysteria1_protocol tests.test_hysteria2_compatibility tests.test_backend_editor_contract tests.test_shadowsocks_uri tests.test_socks_uri -v +python -m unittest tests.test_python_compatibility tests.test_interface tests.test_models_and_services tests.test_repository_contracts tests.test_plugin_architecture tests.test_public_api tests.test_hysteria1_protocol tests.test_hysteria2_compatibility tests.test_backend_editor_contract tests.test_shadowsocks_uri tests.test_socks_uri -v # Application, connection, runtime, and mocked host boundaries python -m unittest tests.test_application_process tests.test_architecture_refactors tests.test_connection_startup_async tests.test_controllers tests.test_runtime_lifecycle tests.test_external_core tests.test_frozenlib tests.test_native_tun_semantics tests.test_sing_tun -v diff --git a/tests/test_python_compatibility.py b/tests/test_python_compatibility.py new file mode 100644 index 00000000..19a7d5a5 --- /dev/null +++ b/tests/test_python_compatibility.py @@ -0,0 +1,397 @@ +# 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 Python compatibility with real workers and legacy API shapes.""" + +from __future__ import annotations + +from Furious.Core.SingTUN import SingTUN +from Furious.Frozenlib import PythonCompatibility +from Furious.Plugins import FuriousPlugin, PluginMetadata, PluginRegistry +from Furious.Service.TrafficStatsManager import TrafficStatsManager + +from shiboken6 import delete as deleteQObject + +from concurrent.futures import ThreadPoolExecutor +from importlib import metadata +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +import threading +import unittest + +from tests.support import ( + application, + assertChildSucceeded, + isolatedSettings, + runPythonChild, +) + + +class _LegacyExecutor: + """Expose the Python 3.8 shutdown signature over a real worker pool.""" + + def __init__(self, executor): + self.executor = executor + + def shutdown(self, wait=True): + self.executor.shutdown(wait=wait) + + +class _MetadataProvider: + """Count capability probes while exposing a strict entry-point API.""" + + PackageNotFoundError = metadata.PackageNotFoundError + + def __init__(self, selectable, entryPoints): + self.selectable = selectable + self.entry_points = entryPoints + self.probes = 0 + + @property + def EntryPoints(self): + self.probes += 1 + + if not self.selectable: + raise AttributeError('legacy metadata has no EntryPoints type') + + return metadata.EntryPoints + + +class PythonCompatibilityTest(unittest.TestCase): + """Preserve exact text, metadata selection, and queued-work cancellation.""" + + @staticmethod + def _loadCompatibility(version, metadataProvider=metadata): + """Execute the checked-out module with isolated interpreter/API inputs.""" + path = ( + Path(__file__).resolve().parents[1] + / 'Furious/Frozenlib/PythonCompatibility.py' + ) + code = compile(path.read_text(encoding='utf-8'), str(path), 'exec') + namespace = {'__name__': '_compatibility_selection'} + + with mock.patch.dict( + 'sys.modules', {'sys': SimpleNamespace(version_info=version)} + ): + with mock.patch('importlib.metadata', metadataProvider): + exec(code, namespace) + + return namespace + + def testExactAffixRemoval(self): + """Remove one match without stripping characters or an empty suffix.""" + cases = ( + ('socks5://localhost:1080', 'socks5://', 'localhost:1080'), + ('socks5://socks5://host', 'socks5://', 'socks5://host'), + ('host', 'socks5://', 'host'), + ('prefix', '', 'prefix'), + ('', 'prefix', ''), + ('prefix', 'prefix', ''), + ('\u524d\u7f00value', '\u524d\u7f00', 'value'), + ) + + for value, prefix, expected in cases: + with self.subTest(value=value, prefix=prefix): + self.assertEqual( + PythonCompatibility.removePrefix(value, prefix), expected + ) + + for value, suffix, expected in ( + ('module.py', '.py', 'module'), + ('module.py.py', '.py', 'module.py'), + ('module', '.py', 'module'), + ('module', '', 'module'), + ('', '.py', ''), + ('.py', '.py', ''), + ('value\u540e\u7f00', '\u540e\u7f00', 'value'), + ): + with self.subTest(value=value, suffix=suffix): + self.assertEqual( + PythonCompatibility.removeSuffix(value, suffix), expected + ) + + for method in ( + PythonCompatibility.removePrefix, + PythonCompatibility.removeSuffix, + ): + with self.assertRaises(TypeError): + method('value', ('value',)) + + def testImplementationsAreSelectedOnceAtImport(self): + """Repeated calls retain their selected APIs without new capability probes.""" + for modern in (False, True): + with self.subTest(modern=modern): + version = mock.MagicMock() + version.__ge__.return_value = modern + fetch = mock.Mock(return_value=() if modern else {}) + provider = _MetadataProvider(modern, fetch) + namespace = self._loadCompatibility(version, provider) + compatibility = namespace['PythonCompatibility'] + + fetch.assert_not_called() + self.assertEqual(provider.probes, 1) + + if modern: + self.assertIs(namespace['_removePrefix'], str.removeprefix) + self.assertIs(namespace['_removeSuffix'], str.removesuffix) + + namespace['_PYTHON_39_OR_NEWER'] = not modern + + for _ in range(10): + self.assertEqual( + compatibility.removePrefix('prefix-value', 'prefix-'), 'value' + ) + self.assertEqual( + compatibility.removeSuffix('value-suffix', '-suffix'), 'value' + ) + self.assertEqual(compatibility.removePrefix('value', ''), 'value') + self.assertEqual(compatibility.removeSuffix('value', ''), 'value') + self.assertEqual(compatibility.entryPoints('missing'), ()) + + executor, future = mock.Mock(), mock.Mock() + compatibility.shutdownExecutor(executor, (future,), wait=False) + + if modern: + executor.shutdown.assert_called_once_with( + wait=False, cancel_futures=True + ) + future.cancel.assert_not_called() + else: + executor.shutdown.assert_called_once_with(wait=False) + future.cancel.assert_called_once_with() + + version.__ge__.assert_called_once_with((3, 9)) + self.assertEqual(provider.probes, 1) + self.assertEqual(fetch.call_count, 10) + + if modern: + fetch.assert_called_with(group='missing') + else: + fetch.assert_called_with() + + def testEntryPointAPIShapesPreserveOrderAndDiscoverPlugins(self): + """Discovery reaches the same registry through old and new metadata APIs.""" + + class FixturePlugin(FuriousPlugin): + metadata = PluginMetadata( + 'tests.python-compatibility', 'Compatibility fixture' + ) + + entry = SimpleNamespace(name='fixture', load=lambda: FixturePlugin) + entries = (entry, SimpleNamespace(name='second', load=lambda: ())) + + for selectable in (False, True): + with self.subTest(selectable=selectable): + if selectable: + + def fetch(*, group): + return entries if group == 'furious.plugins' else () + + else: + + def fetch(): + return {'furious.plugins': entries, 'unrelated': ()} + + provider = _MetadataProvider(selectable, mock.Mock(side_effect=fetch)) + compatibility = self._loadCompatibility((3, 13), provider)[ + 'PythonCompatibility' + ] + + self.assertEqual(compatibility.entryPoints('furious.plugins'), entries) + self.assertEqual(compatibility.entryPoints('missing'), ()) + + registry = PluginRegistry() + try: + with mock.patch( + 'Furious.Plugins.Registry.PythonCompatibility', + compatibility, + ): + registry.discover() + + self.assertEqual( + registry.metadataFor('tests.python-compatibility'), + FixturePlugin.metadata, + ) + finally: + registry.shutdown() + + if selectable: + provider.entry_points.assert_called_with(group='furious.plugins') + else: + provider.entry_points.assert_called_with() + + self.assertEqual(provider.probes, 1) + + def testDistributionVersionKeepsMissingMetadataBehavior(self): + """The engine reports unavailable only for missing distribution metadata.""" + target = 'Furious.Frozenlib.PythonCompatibility.metadata.version' + + with mock.patch(target, return_value='0.9.7.dev0') as version: + self.assertEqual(SingTUN.version(), '0.9.7.dev0') + version.assert_called_once_with('sing-tun') + + with mock.patch( + target, side_effect=PythonCompatibility.PackageNotFoundError('sing-tun') + ): + self.assertEqual(SingTUN.version(), 'unavailable') + + with mock.patch(target, side_effect=ValueError('invalid metadata')): + with self.assertRaisesRegex(ValueError, 'invalid metadata'): + SingTUN.version() + + def testModernShutdownDoesNotRetryUnrelatedTypeErrors(self): + """Native shutdown errors propagate without an unsafe fallback retry.""" + executor = mock.Mock() + executor.shutdown.side_effect = TypeError('shutdown failed') + future = mock.Mock() + + compatibility = self._loadCompatibility((3, 9))['PythonCompatibility'] + + with self.assertRaisesRegex(TypeError, 'shutdown failed'): + compatibility.shutdownExecutor(executor, (future,), wait=False) + + executor.shutdown.assert_called_once_with(wait=False, cancel_futures=True) + future.cancel.assert_not_called() + + def testShutdownCancelsQueuedWorkAndLeavesRunningWorkOwned(self): + """Both APIs reject new work, cancel queued calls, and allow a real drain.""" + for version in ((3, 8), (3, 9)): + with self.subTest(version=version): + started, release, queuedRan = ( + threading.Event(), + threading.Event(), + threading.Event(), + ) + executor = ThreadPoolExecutor(max_workers=1) + + def blocked(): + started.set() + return release.wait(5) + + running = executor.submit(blocked) + queued = None + + try: + self.assertTrue(started.wait(2)) + queued = executor.submit(queuedRan.set) + adapter = ( + _LegacyExecutor(executor) if version == (3, 8) else executor + ) + + compatibility = self._loadCompatibility(version)[ + 'PythonCompatibility' + ] + compatibility.shutdownExecutor(adapter, (queued,), wait=False) + + self.assertTrue(queued.cancelled()) + self.assertFalse(running.done()) + self.assertFalse(queuedRan.is_set()) + + with self.assertRaises(RuntimeError): + executor.submit(lambda: None) + + release.set() + self.assertTrue(running.result(timeout=2)) + finally: + release.set() + if queued is not None: + queued.cancel() + executor.shutdown(wait=True) + + self.assertFalse(queuedRan.is_set()) + self.assertTrue( + all(not thread.is_alive() for thread in executor._threads) + ) + + def testStatisticsOwnerPassesItsPendingWorkToLegacyShutdown(self): + """Manager shutdown cancels its queued query using the strict old API.""" + application() + + with isolatedSettings(): + manager = TrafficStatsManager() + started, release = threading.Event(), threading.Event() + executor = ThreadPoolExecutor(max_workers=1) + + def blocked(): + started.set() + return release.wait(5) + + running = executor.submit(blocked) + queued = None + + try: + self.assertTrue(started.wait(2)) + queued = executor.submit(lambda: None) + manager._executor = _LegacyExecutor(executor) + manager._future = queued + manager._queryInFlight = True + + compatibility = self._loadCompatibility((3, 8))['PythonCompatibility'] + + with mock.patch( + 'Furious.Service.TrafficStatsManager.PythonCompatibility', + compatibility, + ): + manager._closeExecutor() + + self.assertIsNone(manager._executor) + self.assertIsNone(manager._future) + self.assertFalse(manager._queryInFlight) + self.assertTrue(queued.cancelled()) + self.assertFalse(running.done()) + finally: + release.set() + if queued is not None: + queued.cancel() + executor.shutdown(wait=True) + manager.cleanup() + deleteQObject(manager) + + def testProfileColdImportWithAnUnsubscriptableABC(self): + """A Python 3.8-style collections ABC cannot break the profile import.""" + script = """ +from collections.abc import MutableMapping + +def unavailable(cls, item): + raise TypeError('collections ABC is not subscriptable') + +MutableMapping.__class_getitem__ = classmethod(unavailable) + +try: + MutableMapping[str, object] +except TypeError: + pass +else: + raise AssertionError('legacy ABC fixture did not reject subscripting') + +from Furious.Models import CoreConfiguration, ServerProfile + +profile = ServerProfile.fromConfiguration(CoreConfiguration({'server': 'example'})) +assert isinstance(profile, MutableMapping) +assert profile['server'] == 'example' +profile['port'] = 443 +assert profile['port'] == 443 +""" + assertChildSucceeded( + self, runPythonChild(script), 'profile import with a Python 3.8-style ABC' + ) + + +if __name__ == '__main__': + unittest.main()