Merge pull request #523 from f3rs3n/fix/monitor-fixtures-localized-icons

fix(monitor): isolate offline tests and preserve localized field icons

AppImage/scripts/notification_templates.py

@@ -2798,7 +2798,7 @@ def enrich_with_emojis(event_type: str, title: str, body: str,
             label_variants.extend(localized_labels.get(field_key, []))
             
             for label in label_variants:
-                if stripped.lower().startswith(label.lower() + ':'):
+                if re.match(re.escape(label.lower()) + r'[ \u00a0\u202f]*:', stripped.lower()):
                     enriched_lines.append(f'{field_icon} {stripped}')
                     enriched = True
                     break

AppImage/scripts/tests/health_fixture_modules.py

@@ -0,0 +1,60 @@
+"""Private real-module loaders for offline health fixtures.
+
+Never cache test dependencies in sys.modules. Persistence uses its real global
+constructor and SQLite schema, redirected to the caller's temporary directory.
+Monitor methods are unmodified; only unrelated import-time host dependencies and
+singleton cleanup are bounded. Runtime/deferred dependencies belong in each
+calling test's patch.dict context, not in the process-wide module cache.
+"""
+import contextlib
+import importlib.util
+import io
+from pathlib import Path
+import sys
+from types import ModuleType, SimpleNamespace
+from unittest.mock import patch
+
+
+SCRIPTS = Path(__file__).resolve().parents[1]
+# Keep Path usable while the constructor's imported pathlib.Path is redirected.
+ConcretePath = type(Path())
+
+
+def load_source(name):
+    spec = importlib.util.spec_from_file_location('fixture_' + name, SCRIPTS / (name + '.py'))
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    with patch.object(sys, 'dont_write_bytecode', True):
+        spec.loader.exec_module(module)
+    return module
+
+
+def load_health_persistence(data_dir):
+    def fixture_path(*args, **kwargs):
+        if args == ('/usr/local/share/proxmenux',):
+            return ConcretePath(data_dir)
+        return ConcretePath(*args, **kwargs)
+
+    with patch('pathlib.Path', fixture_path), contextlib.redirect_stdout(io.StringIO()):
+        return load_source('health_persistence')
+
+
+def load_health_modules(data_dir):
+    persistence = load_health_persistence(data_dir)
+    # These import-only dependencies are not exercised by these unit tests.
+    # Real methods and the real monitor singleton still execute from source.
+    dependencies = {
+        'health_persistence': persistence,
+        'psutil': ModuleType('psutil'),
+        'proxmox_storage_monitor': SimpleNamespace(proxmox_storage_monitor=SimpleNamespace()),
+        'pbs_manager': SimpleNamespace(get_platform=lambda: {'product': 'pve'}, PBS_SERVICES=()),
+        'startup_grace': ModuleType('startup_grace'),
+    }
+    with patch.dict(sys.modules, dependencies), \
+            patch.object(persistence.health_persistence, 'cleanup_old_errors'), \
+            patch('os.path.exists', return_value=False):
+        monitor = load_source('health_monitor')
+        events = load_source('notification_events')
+    thresholds = load_source('health_thresholds')
+    setattr(thresholds, '_CONFIG_PATH', str(ConcretePath(data_dir) / 'health_thresholds.json'))
+    return monitor, events, thresholds

AppImage/scripts/tests/test_benign_disk_notices.py

@@ -2,12 +2,19 @@ import sys
 from pathlib import Path
 import subprocess
 import unittest
+import tempfile
 from unittest.mock import MagicMock, patch
+from health_fixture_modules import load_health_modules
 
 SCRIPTS = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(SCRIPTS))
-import health_monitor
-import notification_events
+
+
+def setUpModule():
+    global health_monitor, notification_events
+    temporary = tempfile.TemporaryDirectory(prefix='benign-notices-')
+    unittest.addModuleCleanup(temporary.cleanup)
+    health_monitor, notification_events, _ = load_health_modules(temporary.name)
 
 NO_POWER_STATUS = ('Sep 27 09:08:50 proxmox smartd[893]: Device: /dev/sdc [SAT], '
                    'no ATA CHECK POWER STATUS support, ignoring -n Directive')

AppImage/scripts/tests/test_cluster_guest_storage_ownership.py

@@ -2,7 +2,10 @@ import sys
 import unittest
 from pathlib import Path
 from queue import Queue
-from types import ModuleType, SimpleNamespace
+from types import SimpleNamespace
+import tempfile
+
+from health_fixture_modules import load_health_modules
 from unittest.mock import patch
 
 
@@ -10,27 +13,20 @@ SCRIPTS_DIR = Path(__file__).resolve().parents[1]
 if str(SCRIPTS_DIR) not in sys.path:
     sys.path.insert(0, str(SCRIPTS_DIR))
 
-# These modules normally bind the live `/usr/local/share/proxmenux` database
-# while importing. Ownership tests need no host state, so provide the same
-# narrow dependency boundary used by the production functions below.
-_health_persistence_module = ModuleType("health_persistence")
-_health_persistence_module.health_persistence = SimpleNamespace(
-    cleanup_old_errors=lambda: None,
-)
-_health_persistence_module.disk_base_name = lambda name: str(name).replace("/dev/", "")
-sys.modules.setdefault("health_persistence", _health_persistence_module)
-
-sys.modules.setdefault("psutil", ModuleType("psutil"))
+# Dependencies are private to this module's fixture, never published at collection.
 
 flask_server = SimpleNamespace(
     get_proxmox_node_name=lambda: "fixture",
     get_cached_pvesh_cluster_resources_vm=lambda: [],
     get_cached_vm_disk=lambda _vmid: None,
 )
-sys.modules.setdefault("flask_server", flask_server)
 
-import health_monitor  # noqa: E402
-import notification_events  # noqa: E402
+
+def setUpModule():
+    global health_monitor, notification_events, health_thresholds
+    temporary = tempfile.TemporaryDirectory(prefix='cluster-ownership-')
+    unittest.addModuleCleanup(temporary.cleanup)
+    health_monitor, notification_events, health_thresholds = load_health_modules(temporary.name)
 
 
 class _Persistence:
@@ -50,6 +46,11 @@ class _Persistence:
 
 class ClusterGuestStorageOwnershipTests(unittest.TestCase):
     def setUp(self):
+        dependencies = patch.dict(sys.modules, {
+            'flask_server': flask_server, 'health_thresholds': health_thresholds,
+        })
+        dependencies.start()
+        self.addCleanup(dependencies.stop)
         self.monitor = health_monitor.HealthMonitor.__new__(health_monitor.HealthMonitor)
         self.persistence = _Persistence()
         self.resources = [

AppImage/scripts/tests/test_disk_observations_follow_the_disk.py

@@ -1,6 +1,5 @@
 """The observations of a disk belong to its serial number, not to the
 connector it is plugged in."""
-import sqlite3
 import sys
 import tempfile
 import unittest
@@ -9,25 +8,14 @@ from unittest.mock import patch
 
 SCRIPTS = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(SCRIPTS))
-# The module opens the database of the host when it is imported.
-_connect = sqlite3.connect
-with patch.object(Path, 'mkdir'), patch('builtins.print'), \
-        patch('sqlite3.connect', side_effect=lambda *args, **kwargs: _connect(':memory:')):
-    import health_persistence as hp_module
+from health_fixture_modules import load_health_persistence
 
 
 class DiskObservationIdentityTests(unittest.TestCase):
     def setUp(self):
         tmp = tempfile.TemporaryDirectory()
         self.addCleanup(tmp.cleanup)
-        store = hp_module.HealthPersistence.__new__(hp_module.HealthPersistence)
-        for name, value in vars(hp_module.health_persistence).items():
-            setattr(store, name, value)
-        store.data_dir = Path(tmp.name)
-        store.db_path = store.data_dir / 'health_monitor.db'
-        with patch('builtins.print'):
-            store._init_database()
-        self.store = store
+        self.store = load_health_persistence(tmp.name).health_persistence
 
     def record(self, device, serial, signature):
         self.store.register_disk(device, serial)

AppImage/scripts/tests/test_fixture_module_isolation.py

@@ -0,0 +1,91 @@
+"""Test collection must not publish a fixture as a runtime dependency."""
+import importlib.util
+from importlib.machinery import SourceFileLoader
+import tempfile
+from pathlib import Path
+import sys
+from types import ModuleType
+import unittest
+from unittest.mock import patch
+
+
+TESTS = Path(__file__).resolve().parent
+
+
+class FixtureCollectionIsolationTests(unittest.TestCase):
+    def test_private_persistence_loader_never_writes_bytecode(self):
+        from health_fixture_modules import load_health_persistence
+        with tempfile.TemporaryDirectory() as directory, \
+                patch.object(sys, 'dont_write_bytecode', False), \
+                patch.object(sys, 'pycache_prefix', str(Path(directory) / 'empty-cache')), \
+                patch.object(SourceFileLoader, 'set_data') as cache_write:
+            module = load_health_persistence(directory)
+            self.assertEqual(module.health_persistence.db_path,
+                             Path(directory) / 'health_monitor.db')
+            self.assertEqual(cache_write.call_count, 0, 'Fixture imports must not write source bytecode')
+
+    def assert_collection_preserves_modules(self, filename):
+        names = ('health_persistence', 'psutil', 'flask_server', 'health_monitor')
+        # A pre-existing monitor prevents the old fixture from constructing a
+        # host singleton. No runtime method is replaced in the behavioral tests.
+        with patch.dict(sys.modules), patch.object(sys, 'path', list(sys.path)):
+            for name in names:
+                sys.modules.pop(name, None)
+            sys.modules['health_monitor'] = ModuleType('health_monitor')
+            before = {name: sys.modules.get(name) for name in names}
+            spec = importlib.util.spec_from_file_location('collection_fixture', TESTS / filename)
+            assert spec is not None and spec.loader is not None
+            module = importlib.util.module_from_spec(spec)
+            spec.loader.exec_module(module)
+            self.assertEqual({name: sys.modules.get(name) for name in names}, before)
+
+    def assert_collection_defers_runtime_import(self, filename):
+        original_import = __import__
+
+        def fixture_import(name, *args, **kwargs):
+            self.assertNotIn(name, ('health_monitor', 'health_persistence'),
+                             'Collection must not construct host singletons')
+            return original_import(name, *args, **kwargs)
+
+        spec = importlib.util.spec_from_file_location('collection_fixture', TESTS / filename)
+        assert spec is not None and spec.loader is not None
+        module = importlib.util.module_from_spec(spec)
+        with patch('builtins.__import__', side_effect=fixture_import), \
+                patch.object(sys, 'path', list(sys.path)):
+            spec.loader.exec_module(module)
+
+    def test_permanent_cleanup_does_not_open_host_uptime(self):
+        spec = importlib.util.spec_from_file_location(
+            'cleanup_fixture', TESTS / 'test_permanent_dismissal_is_kept.py')
+        assert spec is not None and spec.loader is not None
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        case = module.PermanentDismissalTests('test_it_is_still_dismissed_weeks_later')
+        case.setUp()
+        self.addCleanup(case.doCleanups)
+        with patch('builtins.open', side_effect=AssertionError('No host reads')) as host_open, \
+                patch.dict(sys.modules, {'psutil': ModuleType('psutil')}):
+            case.cleanup()
+        host_open.assert_not_called()
+
+    def test_permanent_dismissal_collection_defers_runtime_import(self):
+        self.assert_collection_defers_runtime_import('test_permanent_dismissal_is_kept.py')
+
+    def test_disk_observation_collection_defers_runtime_import(self):
+        self.assert_collection_defers_runtime_import('test_disk_observations_follow_the_disk.py')
+
+    def test_idle_bridge_collection_defers_runtime_import(self):
+        self.assert_collection_defers_runtime_import('test_idle_bridge.py')
+
+    def test_benign_notice_collection_defers_runtime_import(self):
+        self.assert_collection_defers_runtime_import('test_benign_disk_notices.py')
+
+    def test_remote_mount_collection_does_not_publish_dependency_stubs(self):
+        self.assert_collection_preserves_modules('test_remote_mount_dismissal.py')
+
+    def test_cluster_collection_does_not_publish_dependency_stubs(self):
+        self.assert_collection_preserves_modules('test_cluster_guest_storage_ownership.py')
+
+
+if __name__ == '__main__':
+    unittest.main()

AppImage/scripts/tests/test_health_persistence_recurrence.py

@@ -1,7 +1,7 @@
 """Offline recurrence tests against the complete, real persistence module.
 
 From the repository root (stdlib only; no broad test discovery):
-    python3 -I -B -m unittest discover \
+    python3 -I -m unittest discover \
         -s AppImage/scripts/tests -p test_health_persistence_recurrence.py -v
 
 Each test creates and removes its own private temporary directory.
@@ -31,6 +31,9 @@ from datetime import datetime, timedelta
 from unittest.mock import patch
 
 
+# Keep import bytecode writes outside the offline fixture from tripping the guard.
+sys.dont_write_bytecode = True
+
 # Use the concrete platform class while pathlib.Path is temporarily replaced.
 # Python 3.11's abstract Path.__new__ consults the patched module-level name.
 Path = type(Path())

AppImage/scripts/tests/test_idle_bridge.py

@@ -7,7 +7,15 @@ import unittest
 
 SCRIPTS = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(SCRIPTS))
-from health_monitor import HealthMonitor
+from health_fixture_modules import load_health_modules
+
+
+def setUpModule():
+    global HealthMonitor
+    temporary = tempfile.TemporaryDirectory(prefix='idle-bridge-')
+    unittest.addModuleCleanup(temporary.cleanup)
+    monitor, _, _ = load_health_modules(temporary.name)
+    HealthMonitor = monitor.HealthMonitor
 
 
 class IdleBridgeTests(unittest.TestCase):

AppImage/scripts/tests/test_monitor_links.py

@@ -1,4 +1,5 @@
 import json
+import socket
 from pathlib import Path
 import sys
 import tempfile
@@ -24,6 +25,8 @@ class MonitorLinksTests(unittest.TestCase):
             mock.patch.object(links, "LINKS_FILE", self.root / "links.json"),
             mock.patch.object(links, "SSL_CONFIG_FILE", self.root / "ssl.json"),
             mock.patch.object(links, "STORAGE_CFG", storage),
+            mock.patch.object(links, "_own_addresses", return_value=set()),
+            mock.patch.object(links.socket, "getaddrinfo", side_effect=self.fixture_addresses),
             mock.patch.object(links.pbs_manager, "get_platform", return_value={"product": "pve", "hostname": "amd"}),
             mock.patch.object(links.pbs_manager, "public_servers", return_value=[]),
         ]
@@ -39,6 +42,12 @@ class MonitorLinksTests(unittest.TestCase):
         self.run_patch.start()
         self.addCleanup(self.run_patch.stop)
 
+    @staticmethod
+    def fixture_addresses(host: str, _port):
+        address = {"localhost": "127.0.0.1", "pbs.example.com": "203.0.113.49"}.get(host, host)
+        family = socket.AF_INET6 if ":" in address else socket.AF_INET
+        return [(family, socket.SOCK_STREAM, 6, "", (address, 0))]
+
     def test_normalize_url_keeps_only_the_origin_and_rejects_credentials(self):
         self.assertEqual(links.normalize_url("192.168.0.49:8008/dashboard?x=1"), "https://192.168.0.49:8008")
         self.assertEqual(links.normalize_url("http://pve.lan"), "http://pve.lan")

AppImage/scripts/tests/test_native_app_discovery.py

@@ -533,6 +533,7 @@ class DiscoveryFixtures(unittest.TestCase):
 
     def test_actual_save_first_check_and_recheck_never_probe_or_configure_updater(self):
         with patch.object(lxc_apps, '_APPS_DIR', str(self.root / 'apps')), \
+             patch.object(lxc_apps, '_fetch_tracking_hints', return_value={}), \
              patch.object(lxc_apps, '_pct_exec', side_effect=AssertionError('forbidden guest execution')) as execute, \
              patch.object(lxc_apps, 'fetch_latest_upstream_details', side_effect=AssertionError('forbidden upstream')) as upstream:
             for version in (None, '1.2.3'):

AppImage/scripts/tests/test_notification_delivery_dedup.py

@@ -36,6 +36,9 @@ class NotificationDeliveryDedupTests(unittest.TestCase):
         self.addCleanup(self.temp_dir.cleanup)
         self.db_path = Path(self.temp_dir.name) / 'health_monitor.db'
         conn = sqlite3.connect(str(self.db_path))
+        # Production initializes WAL before dispatchers start; do not race the
+        # journal-mode migration itself in a freshly hand-built DELETE database.
+        conn.execute('PRAGMA journal_mode=WAL')
         conn.execute('''
             CREATE TABLE notification_last_sent (
                 fingerprint TEXT PRIMARY KEY,
@@ -105,6 +108,13 @@ class NotificationDeliveryDedupTests(unittest.TestCase):
             entity_id='210:same-run',
         )
 
+    def test_fixture_uses_the_production_wal_journal_mode(self):
+        connection = sqlite3.connect(str(self.db_path))
+        try:
+            self.assertEqual(connection.execute('PRAGMA journal_mode').fetchone()[0], 'wal')
+        finally:
+            connection.close()
+
     def test_concurrent_managers_deliver_same_fingerprint_once(self):
         channel = RecordingChannel()
         managers = [self._manager(channel), self._manager(channel)]

AppImage/scripts/tests/test_notification_runtime_i18n.py

@@ -227,7 +227,65 @@ class RuntimeCatalogTests(unittest.TestCase):
             total_label = notification_templates._localized_template_labels(
                 "update_summary", language,
             )["total_count"][0]
-            self.assertIn(f"📦 {total_label}: 2", body_locale, language)
+            self.assertRegex(body_locale, rf"(?m)^📦 {re.escape(total_label)}[ \u00a0\u202f]*: 2$", language)
+
+    def test_french_update_icons_preserve_labels_values_and_typography(self):
+        data = {
+            'hostname': 'fixture-pve', 'total_count': '17', 'security_count': '3',
+            'pve_count': '5', 'kernel_count': '7',
+            'important_list': 'fixture-package 1.2.3',
+            'severity': 'INFO', '_notification_language': 'fr',
+        }
+        rendered = notification_templates.render_template('update_summary', data, language='fr')
+        _title, body = notification_templates.enrich_with_emojis(
+            'update_summary', rendered['title'], rendered['body'], data,
+        )
+        for icon, label, value in (
+            ('📦', 'Nombre total de mises à jour', '17'),
+            ('🛡️', 'Mises à jour de sécurité', '3'),
+            ('📦', 'Mises à jour liées à Proxmox', '5'),
+            ('⚙️', 'Mises à jour du noyau', '7'),
+        ):
+            with self.subTest(label=label):
+                self.assertIn(f'{icon} {label}\u00a0: {value}', body.splitlines())
+        self.assertIn('fixture-package 1.2.3', body.splitlines())
+        self.assertNotIn('Total updates:', body)
+
+        versions = {
+            'hostname': 'fixture-pve', 'current_version': '9.2.18',
+            'new_version': '9.2.20', 'details': 'pve-manager 9.2.18 → 9.2.20',
+            'severity': 'INFO', '_notification_language': 'fr',
+        }
+        rendered = notification_templates.render_template('pve_update', versions, language='fr')
+        _title, body = notification_templates.enrich_with_emojis(
+            'pve_update', rendered['title'], rendered['body'], versions,
+        )
+        self.assertIn('Une nouvelle version Proxmox VE est disponible.', body.splitlines())
+        self.assertIn('📦 Actuel\u00a0: 9.2.18', body.splitlines())
+        self.assertIn('🆕 Nouveau\u00a0: 9.2.20', body.splitlines())
+        self.assertIn('🔧 pve-manager 9.2.18 → 9.2.20', body.splitlines())
+
+    def test_field_icons_preserve_horizontal_colon_spacing_and_literal_labels(self):
+        label = 'Résumé (A+B)[0]'
+        labels = {'total_count': [label]}
+        data = {'_notification_language': 'fr', 'severity': 'INFO'}
+        with mock.patch.object(notification_templates, '_localized_template_labels', return_value=labels):
+            for separator in ('', ' ', '\u00a0', '\u202f', ' \u00a0\u202f'):
+                line = f'{label}{separator}: 17'
+                with self.subTest(separator=repr(separator)):
+                    _title, body = notification_templates.enrich_with_emojis(
+                        'update_summary', 'Summary', line, data,
+                    )
+                    self.assertEqual(body, f'📦 {line}')
+            # Prefix collisions, regex-like labels and non-admitted separators
+            # must not gain this field icon. The existing prose branch is separate.
+            for line in (f'{label}X: 17', 'Résumé AAAB0\u00a0: 17',
+                         f'{label}\t: 17', f'{label}\n: 17'):
+                with self.subTest(line=repr(line)):
+                    _title, body = notification_templates.enrich_with_emojis(
+                        'update_summary', 'Summary', line, data,
+                    )
+                    self.assertEqual(body, line)
 
     def test_docker_update_body_icons_preserve_localized_container_label(self):
         data = {
@@ -283,7 +341,7 @@ class RuntimeCatalogTests(unittest.TestCase):
             label = notification_templates._localized_template_labels(
                 "pve_update", language,
             )["new_version"][0]
-            self.assertIn(f"🆕 {label}: 9.2.20", locale_body, language)
+            self.assertRegex(locale_body, rf"(?m)^🆕 {re.escape(label)}[ \u00a0\u202f]*: 9\.2\.20$", language)
 
     def test_missing_slovak_key_falls_back_to_english(self):
         with tempfile.TemporaryDirectory() as directory:

AppImage/scripts/tests/test_pbs.py

@@ -82,6 +82,17 @@ class PBSManagerTests(unittest.TestCase):
         temporary = tempfile.TemporaryDirectory()
         self.addCleanup(temporary.cleanup)
         self.root = Path(temporary.name)
+        import monitor_links
+        storage = self.root / 'storage.cfg'
+        storage.write_text('')
+        for boundary in (
+            mock.patch.object(pbs, 'get_platform', return_value={'product': 'pve', 'hostname': 'fixture'}),
+            mock.patch.object(pbs.shutil, 'which', return_value=None),
+            mock.patch.object(monitor_links, 'STORAGE_CFG', storage),
+            mock.patch.object(monitor_links, '_own_addresses', return_value=set()),
+        ):
+            boundary.start()
+            self.addCleanup(boundary.stop)
         patch = mock.patch.multiple(auth_manager, CONFIG_DIR=self.root, AUTH_CONFIG_FILE=self.root / "auth.json")
         patch.start(); self.addCleanup(patch.stop)
         auth_manager.save_auth_config({"enabled": True, "username": "admin", "password_hash": "configured",

AppImage/scripts/tests/test_permanent_dismissal_is_kept.py

@@ -1,21 +1,17 @@
 """An alert dismissed permanently stays dismissed until the user re-enables
 it: not seeing it for a while, the condition recovering or a change of the
 suppression settings does not bring it back."""
-import sqlite3
 import sys
 import tempfile
 import unittest
 from datetime import datetime, timedelta
 from pathlib import Path
+from types import ModuleType
 from unittest.mock import patch
 
 SCRIPTS = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(SCRIPTS))
-# The module opens the database of the host when it is imported.
-_connect = sqlite3.connect
-with patch.object(Path, 'mkdir'), patch('builtins.print'), \
-        patch('sqlite3.connect', side_effect=lambda *args, **kwargs: _connect(':memory:')):
-    import health_persistence as hp_module
+from health_fixture_modules import load_health_persistence
 
 KEY = 'pve_storage_full_local-lvm'
 
@@ -24,14 +20,8 @@ class PermanentDismissalTests(unittest.TestCase):
     def setUp(self):
         tmp = tempfile.TemporaryDirectory()
         self.addCleanup(tmp.cleanup)
-        store = hp_module.HealthPersistence.__new__(hp_module.HealthPersistence)
-        for name, value in vars(hp_module.health_persistence).items():
-            setattr(store, name, value)
-        store.data_dir = Path(tmp.name)
-        store.db_path = store.data_dir / 'health_monitor.db'
-        with patch('builtins.print'):
-            store._init_database()
-        self.store = store
+        self.module = load_health_persistence(tmp.name)
+        self.store = self.module.health_persistence
 
     def seen(self):
         return self.store.record_error(KEY, 'storage', 'WARNING', 'PVE storage local-lvm (lvmthin) at 96.0% used',
@@ -50,7 +40,10 @@ class PermanentDismissalTests(unittest.TestCase):
         conn.close()
 
     def cleanup(self):
-        with patch('builtins.print'), patch('subprocess.run', side_effect=OSError('no host tools in the test')):
+        with patch('builtins.print'), \
+                patch('subprocess.run', side_effect=OSError('no host tools in the test')), \
+                patch.dict(sys.modules, {'psutil': ModuleType('psutil')}), \
+                patch.object(self.module, 'open', side_effect=OSError('no host probes in the test'), create=True):
             self.store.cleanup_old_errors()
 
     def test_it_is_still_dismissed_weeks_later(self):

AppImage/scripts/tests/test_pve_notifications_cfg_writes.py

@@ -2,6 +2,8 @@
 mounted cluster filesystem, and only when their content has to change."""
 import ast
 import os
+import sys
+from types import SimpleNamespace
 import tempfile
 import unittest
 from pathlib import Path
@@ -53,6 +55,11 @@ class PveNotificationWriteTests(unittest.TestCase):
         self.addCleanup(tmp.cleanup)
         self.pve = Path(tmp.name) / 'pve'
         self.pve.mkdir()
+        platform = mock.patch.dict(sys.modules, {
+            'pbs_manager': SimpleNamespace(is_pbs_host=lambda: False),
+        })
+        platform.start()
+        self.addCleanup(platform.stop)
         self.scope = load(self.pve)
         self.kept = Path(self.scope['_PVE_BACKUP_DIR'])
         self.mounted = True

AppImage/scripts/tests/test_remote_mount_dismissal.py

@@ -2,25 +2,20 @@
 import sys
 import unittest
 from pathlib import Path
-from types import ModuleType, SimpleNamespace
+from types import SimpleNamespace
+import tempfile
+from health_fixture_modules import load_health_modules
 from unittest.mock import patch
 
 SCRIPTS_DIR = Path(__file__).resolve().parents[1]
 if str(SCRIPTS_DIR) not in sys.path:
     sys.path.insert(0, str(SCRIPTS_DIR))
 
-_health_persistence_module = ModuleType("health_persistence")
-_health_persistence_module.health_persistence = SimpleNamespace(cleanup_old_errors=lambda: None)
-_health_persistence_module.disk_base_name = lambda name: str(name).replace("/dev/", "")
-sys.modules.setdefault("health_persistence", _health_persistence_module)
-sys.modules.setdefault("psutil", ModuleType("psutil"))
-sys.modules.setdefault("flask_server", SimpleNamespace(
-    get_proxmox_node_name=lambda: "fixture",
-    get_cached_pvesh_cluster_resources_vm=lambda: [],
-    get_cached_vm_disk=lambda _vmid: None,
-))
-
-import health_monitor  # noqa: E402
+def setUpModule():
+    global health_monitor
+    temporary = tempfile.TemporaryDirectory(prefix='remote-mount-')
+    unittest.addModuleCleanup(temporary.cleanup)
+    health_monitor, _, _ = load_health_modules(temporary.name)
 
 
 class _Persistence: