diff --git a/AppImage/scripts/notification_templates.py b/AppImage/scripts/notification_templates.py index d0d87097..473852e9 100644 --- a/AppImage/scripts/notification_templates.py +++ b/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 diff --git a/AppImage/scripts/tests/health_fixture_modules.py b/AppImage/scripts/tests/health_fixture_modules.py new file mode 100644 index 00000000..ce6e4fbb --- /dev/null +++ b/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 diff --git a/AppImage/scripts/tests/test_benign_disk_notices.py b/AppImage/scripts/tests/test_benign_disk_notices.py index 00c6944f..35401222 100644 --- a/AppImage/scripts/tests/test_benign_disk_notices.py +++ b/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') diff --git a/AppImage/scripts/tests/test_cluster_guest_storage_ownership.py b/AppImage/scripts/tests/test_cluster_guest_storage_ownership.py index 8d49b380..6842b827 100644 --- a/AppImage/scripts/tests/test_cluster_guest_storage_ownership.py +++ b/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 = [ diff --git a/AppImage/scripts/tests/test_disk_observations_follow_the_disk.py b/AppImage/scripts/tests/test_disk_observations_follow_the_disk.py index c2a8f9d7..2e445046 100644 --- a/AppImage/scripts/tests/test_disk_observations_follow_the_disk.py +++ b/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) diff --git a/AppImage/scripts/tests/test_fixture_module_isolation.py b/AppImage/scripts/tests/test_fixture_module_isolation.py new file mode 100644 index 00000000..99a354a6 --- /dev/null +++ b/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() diff --git a/AppImage/scripts/tests/test_health_persistence_recurrence.py b/AppImage/scripts/tests/test_health_persistence_recurrence.py index 8c3c78ab..ebbf4490 100644 --- a/AppImage/scripts/tests/test_health_persistence_recurrence.py +++ b/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()) diff --git a/AppImage/scripts/tests/test_idle_bridge.py b/AppImage/scripts/tests/test_idle_bridge.py index 9fedb568..a3e70158 100644 --- a/AppImage/scripts/tests/test_idle_bridge.py +++ b/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): diff --git a/AppImage/scripts/tests/test_monitor_links.py b/AppImage/scripts/tests/test_monitor_links.py index 1c652f4b..bc6bafea 100644 --- a/AppImage/scripts/tests/test_monitor_links.py +++ b/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") diff --git a/AppImage/scripts/tests/test_native_app_discovery.py b/AppImage/scripts/tests/test_native_app_discovery.py index 5c89ed16..6c7cd493 100644 --- a/AppImage/scripts/tests/test_native_app_discovery.py +++ b/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'): diff --git a/AppImage/scripts/tests/test_notification_delivery_dedup.py b/AppImage/scripts/tests/test_notification_delivery_dedup.py index a88d6675..d04b1ca4 100644 --- a/AppImage/scripts/tests/test_notification_delivery_dedup.py +++ b/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)] diff --git a/AppImage/scripts/tests/test_notification_runtime_i18n.py b/AppImage/scripts/tests/test_notification_runtime_i18n.py index 58961d09..f76079e3 100644 --- a/AppImage/scripts/tests/test_notification_runtime_i18n.py +++ b/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: diff --git a/AppImage/scripts/tests/test_pbs.py b/AppImage/scripts/tests/test_pbs.py index e86c38ac..67eccfbd 100644 --- a/AppImage/scripts/tests/test_pbs.py +++ b/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", diff --git a/AppImage/scripts/tests/test_permanent_dismissal_is_kept.py b/AppImage/scripts/tests/test_permanent_dismissal_is_kept.py index 1da297bc..58c883f5 100644 --- a/AppImage/scripts/tests/test_permanent_dismissal_is_kept.py +++ b/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): diff --git a/AppImage/scripts/tests/test_pve_notifications_cfg_writes.py b/AppImage/scripts/tests/test_pve_notifications_cfg_writes.py index b1ec8e35..339fcb24 100644 --- a/AppImage/scripts/tests/test_pve_notifications_cfg_writes.py +++ b/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 diff --git a/AppImage/scripts/tests/test_remote_mount_dismissal.py b/AppImage/scripts/tests/test_remote_mount_dismissal.py index c66ccf17..73c29ee4 100644 --- a/AppImage/scripts/tests/test_remote_mount_dismissal.py +++ b/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: