diff --git a/.github/scripts/run_offline_qualified.py b/.github/scripts/run_offline_qualified.py index fd2bebfe..9bd13abc 100644 --- a/.github/scripts/run_offline_qualified.py +++ b/.github/scripts/run_offline_qualified.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Run only the independently qualified, host-independent tests/ fixtures. +"""Run only the independently qualified, host-independent fixtures. No filesystem discovery widens this manifest. Run from the repository root; Python and Node have separate CI jobs so either failure blocks the check. @@ -10,6 +10,7 @@ import os from pathlib import Path import subprocess import sys +import tempfile PYTHON_FILES = ( "tests/test_audit_presentation.py", @@ -20,6 +21,7 @@ PYTHON_FILES = ( "tests/storage/test_nvme_status_message.py", "tests/test_fastfetch_config_generation.py", "tests/test_notification_active_payloads.py", + "AppImage/scripts/tests/test_health_persistence_recurrence.py", ) NODE_FILES = ( "tests/lxc_updates/test_docker_delegated_ui.cjs", @@ -69,9 +71,13 @@ def main() -> int: if args.lane == "python": folder, filename = str(Path(path).parent), Path(path).name command = [sys.executable, "-I", "-B", "-c", PYTHON_RUN, folder, filename] + with tempfile.TemporaryDirectory(prefix="qualified-offline-") as scratch: + environment = os.environ.copy() + environment["TMPDIR"] = scratch + result = subprocess.run(command, cwd=root, env=environment, check=False) else: command = ["node", path] - result = subprocess.run(command, cwd=root, env=os.environ.copy(), check=False) + result = subprocess.run(command, cwd=root, env=os.environ.copy(), check=False) if result.returncode != 0: print(f"FAIL {path}: exit {result.returncode}", file=sys.stderr) return 1 diff --git a/.github/workflows/test-offline-qualified.yml b/.github/workflows/test-offline-qualified.yml index df57a839..d16ed340 100644 --- a/.github/workflows/test-offline-qualified.yml +++ b/.github/workflows/test-offline-qualified.yml @@ -11,6 +11,7 @@ on: - 'AppImage/package.json' - 'AppImage/package-lock.json' - 'AppImage/scripts/*.py' + - 'AppImage/scripts/tests/test_health_persistence_recurrence.py' - 'AppImage/app/**' - 'AppImage/components/**' - 'AppImage/hooks/**' @@ -30,6 +31,7 @@ on: - 'AppImage/package.json' - 'AppImage/package-lock.json' - 'AppImage/scripts/*.py' + - 'AppImage/scripts/tests/test_health_persistence_recurrence.py' - 'AppImage/app/**' - 'AppImage/components/**' - 'AppImage/hooks/**' @@ -54,7 +56,7 @@ jobs: - uses: actions/setup-python@v5 with: python-version: '3.11' - - name: Run eight qualified Python files + - name: Run nine qualified Python files env: PYTHONDONTWRITEBYTECODE: '1' run: python3 -I -B .github/scripts/run_offline_qualified.py --lane python diff --git a/AppImage/scripts/health_persistence.py b/AppImage/scripts/health_persistence.py index 6230aec9..e8062e42 100644 --- a/AppImage/scripts/health_persistence.py +++ b/AppImage/scripts/health_persistence.py @@ -630,6 +630,12 @@ class HealthPersistence: event_info = {'type': 'updated', 'needs_notification': False} + if existing and existing[1] == 0 and existing[2] is not None: + # A resolved, non-dismissed condition returning is a new episode. + # Keep event history and reuse normal insertion/suppression policy. + cursor.execute('DELETE FROM errors WHERE id = ?', (existing[0],)) + existing = None + if existing: (err_id, ack, resolved_at, old_cat, old_severity, first_seen, notif_sent, stored_suppression, acknowledged_at) = existing diff --git a/AppImage/scripts/tests/test_health_persistence_recurrence.py b/AppImage/scripts/tests/test_health_persistence_recurrence.py new file mode 100644 index 00000000..651d1e06 --- /dev/null +++ b/AppImage/scripts/tests/test_health_persistence_recurrence.py @@ -0,0 +1,495 @@ +"""Offline recurrence tests against the complete, real persistence module. + +From the repository root (stdlib only; no broad test discovery): + TMPDIR=/path/to/private/scratch python3 -I -B -m unittest discover \ + -s AppImage/scripts/tests -p test_health_persistence_recurrence.py -v + +TMPDIR must already exist and must not be the system temporary directory. +Only the data-directory Path boundary and datetime clock are redirected, before +module execution (including the real global constructor). No persistence method +is stubbed. Every test has its own real SQLite DB. Audit guards constrain DB +connections and file opens, restrict mkdir/remove/rmdir/rename to the fixture, +and reject subprocess/socket operations. Metadata/existence probes (for example +os.path.exists) are not comprehensively guarded; synthetic keys avoid VM/disk +resource-existence checks. Read opens are limited to fixture/source/stdlib paths +except the nonexistent synthetic traceback filename described in the guard. +These are synthetic recorder tests, not transport, UI or deployment validation. +""" + +import contextlib +import importlib.util +import io +import json +import os +from pathlib import Path +import sqlite3 +import sys +import tempfile +import unittest +from datetime import datetime, timedelta +from unittest.mock import patch + + +# 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()) +ROOT = Path(__file__).resolve().parents[3] +SOURCE = ROOT / 'AppImage/scripts/health_persistence.py' +_AUDIT_SCOPE = None + + +class BoundaryViolation(RuntimeError): + pass + + +def _audit_guard(event, args): + scope = _AUDIT_SCOPE + if scope is None: + return + + def reject(): + scope['blocked'].append((event, repr(args))) + raise BoundaryViolation('Offline fixture rejected ' + event) + + def inside(path, directory): + return Path(os.fsdecode(path)).resolve().is_relative_to(directory) + + if event.startswith(('socket.', 'subprocess.', 'os.system', 'os.exec', + 'os.spawn', 'os.fork', 'pty.')): + reject() + if event == 'sqlite3.connect': + if Path(os.fsdecode(args[0])).resolve() != scope['db']: + reject() + scope['connections'].append(str(args[0])) + if event == 'open': + path, mode, flags = args + if isinstance(path, int): + reject() + writing = (isinstance(mode, str) and any(c in mode for c in 'wax+')) or ( + isinstance(flags, int) and flags & ( + os.O_WRONLY | os.O_RDWR | os.O_CREAT | os.O_TRUNC)) + if writing: + if not inside(path, scope['root']): + reject() + elif not any(inside(path, directory) for directory in ( + scope['root'], ROOT, Path(sys.base_prefix) / 'lib')): + # unittest traceback formatting may ask tokenize to read this exact + # synthetic name. Permit only the harmless missing-file read; an + # existing file or even a dangling symlink must still be rejected. + if not (path == '' and mode == 'rb' and flags == 0 + and not os.path.lexists(path)): + reject() + if event in ('os.mkdir', 'os.remove', 'os.rmdir', 'os.rename'): + paths = args[:2] if event == 'os.rename' else args[:1] + if not all(inside(path, scope['root']) for path in paths): + reject() + + +sys.addaudithook(_audit_guard) + + +class FrozenDatetime(datetime): + current = datetime(2026, 10, 1, 12) + + @classmethod + def now(cls, tz=None): + if tz is not None: + raise AssertionError('Unexpected timezone request') + return cls.current + + +class HealthPersistenceRecurrenceTests(unittest.TestCase): + def setUp(self): + global _AUDIT_SCOPE + scratch = os.environ.get('TMPDIR') + if not scratch or not Path(scratch).is_dir(): + self.fail('Set TMPDIR to an existing private scratch directory') + if Path(scratch).resolve() in (Path('/tmp'), Path('/var/tmp')): + self.fail('System temporary directories are not allowed') + temporary = tempfile.TemporaryDirectory(prefix='health-recurrence-', dir=scratch) + self.addCleanup(temporary.cleanup) + self.data_dir = Path(temporary.name) + self.scope = {'root': self.data_dir, 'db': self.data_dir / 'health_monitor.db', + 'blocked': [], 'connections': []} + _AUDIT_SCOPE = self.scope + self.addCleanup(self.finish_isolation) + FrozenDatetime.current = datetime(2026, 10, 1, 12) + + def redirected_path(*args, **kwargs): + if args == ('/usr/local/share/proxmenux',): + return self.data_dir + return Path(*args, **kwargs) + + spec = importlib.util.spec_from_file_location('offline_health_persistence', SOURCE) + assert spec is not None and spec.loader is not None + self.module = importlib.util.module_from_spec(spec) + with patch('pathlib.Path', redirected_path), \ + patch('datetime.datetime', FrozenDatetime), \ + contextlib.redirect_stdout(io.StringIO()): + spec.loader.exec_module(self.module) + self.db = self.module.health_persistence + self.assertEqual(self.db.db_path, self.scope['db']) + + def finish_isolation(self): + global _AUDIT_SCOPE + _AUDIT_SCOPE = None + self.assertEqual(self.scope['blocked'], [], 'Even swallowed boundary violations fail') + self.assertTrue(self.scope['connections'], 'The real SQLite constructor must run') + + def advance(self, **kwargs): + FrozenDatetime.current += timedelta(**kwargs) + + def record(self, key='security_updates', category='updates', severity='INFO', + reason='Synthetic condition', details=None): + return self.db.record_error(key, category, severity, reason, details) + + def snapshot(self): + # Independent fresh SQLite connection verifies durable, committed state. + conn = sqlite3.connect(str(self.db.db_path)) + conn.row_factory = sqlite3.Row + try: + rows = [dict(row) for row in conn.execute('SELECT * FROM errors ORDER BY id')] + events = [dict(row) for row in conn.execute('SELECT * FROM events ORDER BY id')] + for event in events: + event['data'] = json.loads(event['data']) + return rows, events + finally: + conn.close() + + def test_notified_resolved_condition_returns_as_a_fresh_episode(self): + key = 'security_updates' + self.assertEqual(self.record(), {'type': 'new', 'needs_notification': True}) + self.db.mark_notified(key) + self.advance(minutes=1) + self.db.resolve_error(key, 'Synthetic condition absent') + self.assertEqual(self.db.get_active_errors(), []) + self.assertEqual(self.db.get_unnotified_errors(), []) + self.advance(minutes=1) + details = {'name': 'offline fixture', 'synthetic': True, 'pending': 3} + result = self.record(reason='Synthetic condition returned', details=details) + rows, events = self.snapshot() + self.assertIsNone(rows[0]['resolved_at'], 'Resolved condition must reactivate') + self.assertEqual(result, {'type': 'new', 'needs_notification': True}) + row = rows[0] + self.assertEqual(len(rows), 1) + self.assertEqual(row['first_seen'], FrozenDatetime.current.isoformat()) + self.assertEqual(row['last_seen'], FrozenDatetime.current.isoformat()) + self.assertEqual(row['notification_sent'], 0) + self.assertEqual(row['acknowledged'], 0) + self.assertIsNone(row['resolution_type']) + self.assertIsNone(row['resolution_reason']) + self.assertEqual(json.loads(row['details']), details) + self.assertTrue(self.db.is_error_active(key, 'updates')) + self.assertEqual(self.db.get_error_by_key(key)['reason'], 'Synthetic condition returned') + self.assertEqual([e['error_key'] for e in self.db.get_active_errors('updates')], [key]) + self.assertEqual([e['error_key'] for e in self.db.get_unnotified_errors()], [key]) + self.assertEqual([event['event_type'] for event in events], ['new', 'resolved', 'new']) + self.assertEqual(events[-1]['data']['details'], details) + self.assertEqual(events[-1]['data']['entity'], 'offline fixture') + + def test_resolve_and_clear_recurrence_matrix(self): + for severity in ('INFO', 'WARNING', 'CRITICAL'): + for route in ('resolve_error', 'clear_error'): + for notified in (False, True): + key = f'fixture_{severity}_{route}_{notified}' + with self.subTest(severity=severity, route=route, notified=notified): + self.record(key=key, severity=severity) + if notified: + self.db.mark_notified(key) + self.advance(minutes=1) + getattr(self.db, route)(key) + self.assertFalse(self.db.is_error_active(key)) + self.advance(minutes=1) + self.assertEqual(self.record(key=key, severity=severity), + {'type': 'new', 'needs_notification': True}) + row = self.db.get_error_by_key(key) + self.assertIsNotNone(row) + for field in ('resolved_at', 'resolution_type', 'resolution_reason', + 'acknowledged_at'): + self.assertIsNone(row[field], field) + self.assertEqual(row['notification_sent'], 0) + self.assertEqual(row['acknowledged'], 0) + self.assertEqual(row['first_seen'], FrozenDatetime.current.isoformat()) + self.assertEqual(row['last_seen'], row['first_seen']) + self.assertIn(key, [e['error_key'] for e in self.db.get_unnotified_errors()]) + rows, events = self.snapshot() + self.assertEqual(sum(row['error_key'] == key for row in rows), 1) + self.assertEqual([e['event_type'] for e in events if e['error_key'] == key], + ['new', 'resolved' if route == 'resolve_error' else 'cleared', 'new']) + + def test_repeated_polls_after_reopening_rearm_only_once(self): + key = 'security_updates' + self.record() + self.db.mark_notified(key) + self.db.resolve_error(key) + self.advance(minutes=1) + self.assertEqual(self.record(), {'type': 'new', 'needs_notification': True}) + first_seen = self.db.get_error_by_key(key)['first_seen'] + for poll in range(4): + self.advance(minutes=1) + self.assertEqual(self.record(), {'type': 'updated', 'needs_notification': False}) + self.assertEqual(self.db.get_error_by_key(key)['first_seen'], first_seen) + self.assertEqual(self.db.get_error_by_key(key)['notification_sent'], int(poll > 0)) + if poll == 0: + self.db.mark_notified(key) + rows, events = self.snapshot() + self.assertEqual(len(rows), 1) + self.assertEqual([e['event_type'] for e in events], + ['new', 'resolved', 'new'] + ['updated'] * 4) + self.assertEqual(self.db.get_unnotified_errors(), []) + # A second real constructor sees exactly the same committed episode/history. + with contextlib.redirect_stdout(io.StringIO()): + self.db = self.module.HealthPersistence() + self.assertEqual(self.snapshot(), (rows, events)) + self.assertTrue(self.db.is_error_active(key)) + self.assertEqual(self.db.get_unnotified_errors(), []) + + def test_same_key_can_resolve_and_recur_twice(self): + key = 'security_updates' + self.record() + for _ in range(2): + self.db.mark_notified(key) + self.advance(minutes=1) + self.db.resolve_error(key, 'Synthetic recovery') + resolved = self.snapshot() + self.db.resolve_error(key, 'Duplicate recovery poll') + self.assertEqual(self.snapshot(), resolved) + self.advance(minutes=1) + self.assertEqual(self.record(), {'type': 'new', 'needs_notification': True}) + self.assertEqual(self.db.get_error_by_key(key)['notification_sent'], 0) + rows, events = self.snapshot() + self.assertEqual(len(rows), 1) + self.assertEqual([e['event_type'] for e in events], + ['new', 'resolved', 'new', 'resolved', 'new']) + self.assertEqual([e['data']['reason'] for e in events if e['event_type'] == 'resolved'], + ['Synthetic recovery', 'Synthetic recovery']) + + def test_active_polls_preserve_existing_episode_and_notification_state(self): + for severity in ('INFO', 'WARNING'): + for notified in (False, True): + key = f'fixture_active_{severity}_{notified}' + with self.subTest(severity=severity, notified=notified): + self.record(key=key, severity=severity) + if notified: + self.db.mark_notified(key) + before = self.db.get_error_by_key(key) + self.advance(minutes=1) + self.assertEqual(self.record(key=key, severity=severity, reason='Next poll'), + {'type': 'updated', 'needs_notification': False}) + after = self.db.get_error_by_key(key) + for field in ('first_seen', 'notification_sent', 'acknowledged', + 'resolved_at', 'resolution_type', 'resolution_reason'): + self.assertEqual(after[field], before[field], field) + self.assertEqual(after['last_seen'], FrozenDatetime.current.isoformat()) + self.assertEqual(after['reason'], 'Next poll') + + def test_active_warning_escalation_is_unchanged(self): + self.record(severity='WARNING') + self.db.mark_notified('security_updates') + before = self.db.get_error_by_key('security_updates') + self.advance(minutes=1) + self.assertEqual(self.record(severity='CRITICAL'), + {'type': 'escalated', 'needs_notification': True}) + after = self.db.get_error_by_key('security_updates') + self.assertEqual(after['first_seen'], before['first_seen']) + self.assertEqual(after['notification_sent'], 1) + self.assertIsNone(after['resolved_at']) + + def test_acknowledged_rows_within_window_are_untouched(self): + for resolved in (False, True): + key = f'fixture_timed_{resolved}' + with self.subTest(resolved=resolved): + self.record(key=key, severity='WARNING') + self.db.mark_notified(key) + self.db.acknowledge_error(key, suppression_hours=24) + if resolved: + self.db.resolve_error(key) + before = self.snapshot() + self.advance(hours=23) + self.assertEqual(self.record(key=key, severity='CRITICAL'), + {'type': 'skipped_acknowledged', 'needs_notification': False}) + self.assertEqual(self.snapshot(), before) + self.assertTrue(self.db.is_error_acknowledged(key)) + self.assertTrue(self.db.is_error_active(key)) + + def test_permanent_dismissal_survives_recovery_and_recurrence(self): + for resolved_before_dismiss in (False, True): + key = f'fixture_permanent_{resolved_before_dismiss}' + with self.subTest(resolved_before_dismiss=resolved_before_dismiss): + self.db.set_setting('suppress_updates', '24') + self.record(key=key, severity='WARNING') + self.db.mark_notified(key) + if resolved_before_dismiss: + self.db.resolve_error(key) + self.db.acknowledge_error(key, suppression_hours=-1) + before = self.snapshot() + self.db.clear_error(key) + self.db.resolve_error(key) + self.advance(days=30) + self.db.set_setting('suppress_updates', '48') + self.assertEqual(self.record(key=key, severity='CRITICAL'), + {'type': 'skipped_acknowledged', 'needs_notification': False}) + self.assertEqual(self.snapshot(), before) + self.assertTrue(self.db.is_error_permanently_acknowledged(key)) + self.assertEqual(self.db.get_active_errors(), []) + self.assertEqual(self.db.get_unnotified_errors(), []) + + def test_expired_resolved_dismissal_rearms_once(self): + key = 'security_updates' + self.record(severity='WARNING') + self.db.mark_notified(key) + self.db.acknowledge_error(key, suppression_hours=24) + self.db.resolve_error(key) + self.advance(hours=25) + self.assertEqual(self.record(severity='WARNING'), + {'type': 'new', 'needs_notification': True}) + row = self.db.get_error_by_key(key) + self.assertEqual(row['first_seen'], FrozenDatetime.current.isoformat()) + self.assertIsNone(row['resolved_at']) + self.assertEqual(row['acknowledged'], 0) + self.assertEqual(row['notification_sent'], 0) + self.assertEqual(self.record(severity='WARNING'), + {'type': 'updated', 'needs_notification': False}) + self.assertEqual([e['event_type'] for e in self.snapshot()[1]], + ['new', 'acknowledged', 'resolved', 'new', 'updated']) + + def test_recurrence_uses_current_category_auto_suppression(self): + for category, setting in (('updates', 'suppress_updates'), ('memory', 'suppress_memory')): + for hours in (48, -1): + key = f'fixture_auto_{category}_{hours}' + with self.subTest(category=category, hours=hours): + # Establish each case independently, even if an earlier + # subtest failed before it could restore shared settings. + self.db.set_setting(setting, '24') + self.record(key=key, category=category, severity='WARNING') + self.db.mark_notified(key) + self.db.resolve_error(key) + self.db.set_setting(setting, str(hours)) + self.advance(minutes=1) + self.assertEqual(self.record(key=key, category=category, severity='WARNING'), + {'type': 'auto_suppressed', 'needs_notification': False}) + rows, events = self.snapshot() + row = next(row for row in rows if row['error_key'] == key) + self.assertIsNone(row['resolved_at']) + self.assertIsNone(row['resolution_type']) + self.assertIsNone(row['resolution_reason']) + self.assertEqual(row['notification_sent'], 0) + self.assertEqual(row['acknowledged'], 1) + self.assertEqual(row['suppression_hours'], hours) + self.assertEqual(row['first_seen'], FrozenDatetime.current.isoformat()) + self.assertEqual(row['acknowledged_at'], row['first_seen']) + self.assertEqual([e['event_type'] for e in events if e['error_key'] == key], + ['new', 'resolved', 'auto_suppressed']) + before = self.snapshot() + self.assertEqual(self.record(key=key, category=category, severity='WARNING'), + {'type': 'skipped_acknowledged', 'needs_notification': False}) + self.assertEqual(self.snapshot(), before) + self.assertNotIn(key, [e['error_key'] for e in self.db.get_active_errors()]) + self.assertNotIn(key, [e['error_key'] for e in self.db.get_unnotified_errors()]) + self.db.set_setting(setting, '24') + + def test_critical_cpu_recurrence_bypasses_category_auto_suppression(self): + key = 'cpu_temperature' + self.record(key=key, category='temperature', severity='WARNING') + self.db.mark_notified(key) + self.db.resolve_error(key) + self.db.set_setting('suppress_cpu', '-1') + self.advance(minutes=1) + self.assertEqual(self.record(key=key, category='temperature', severity='CRITICAL'), + {'type': 'new', 'needs_notification': True}) + row = self.db.get_error_by_key(key) + self.assertEqual(row['acknowledged'], 0) + self.assertEqual(row['notification_sent'], 0) + self.assertIsNone(row['resolved_at']) + self.assertEqual(self.record(key=key, category='temperature', severity='CRITICAL'), + {'type': 'updated', 'needs_notification': False}) + + def test_critical_cpu_override_of_permanent_dismissal_is_unchanged(self): + key = 'cpu_temperature' + self.record(key=key, category='temperature', severity='WARNING') + self.db.acknowledge_error(key, suppression_hours=-1) + self.db.clear_error(key) + self.db.resolve_error(key) + self.db.set_setting('suppress_cpu', '-1') + self.advance(minutes=1) + self.assertEqual(self.record(key=key, category='temperature', severity='CRITICAL'), + {'type': 'new', 'needs_notification': True}) + row = self.db.get_error_by_key(key) + self.assertEqual(row['acknowledged'], 0) + self.assertEqual(row['notification_sent'], 0) + self.assertIsNone(row['resolved_at']) + self.assertEqual([e['event_type'] for e in self.snapshot()[1]], + ['new', 'acknowledged', 'new']) + + def test_reopening_one_key_preserves_other_rows_and_delivery_state(self): + self.record() + self.record(key='security_updates_other', severity='WARNING') + self.db.mark_notified('security_updates_other') + self.db.resolve_error('security_updates') + other = self.db.get_error_by_key('security_updates_other') + conn = sqlite3.connect(str(self.db.db_path)) + try: + conn.execute('INSERT INTO notification_delivery_claims VALUES (?, ?, ?)', + ('health_security_updates', 'in-flight-token', 123)) + conn.execute('INSERT INTO notification_last_sent VALUES (?, ?, ?)', + ('health_security_updates', 123, 1)) + conn.commit() + self.advance(minutes=1) + self.record() + self.assertEqual(self.db.get_error_by_key('security_updates_other'), other) + self.assertEqual(conn.execute('SELECT * FROM notification_delivery_claims').fetchall(), + [('health_security_updates', 'in-flight-token', 123)]) + self.assertEqual(conn.execute('SELECT * FROM notification_last_sent').fetchall(), + [('health_security_updates', 123, 1)]) + finally: + conn.close() + + def test_failed_reopening_rolls_back_without_losing_resolved_episode(self): + self.record() + self.db.mark_notified('security_updates') + self.db.resolve_error('security_updates') + before = self.snapshot() + conn = sqlite3.connect(str(self.db.db_path)) + try: + conn.execute('''CREATE TRIGGER fixture_reject_new_episode BEFORE INSERT ON errors + BEGIN SELECT RAISE(ABORT, 'fixture rejects insert'); END''') + conn.commit() + finally: + conn.close() + self.advance(minutes=1) + with self.assertRaisesRegex(sqlite3.IntegrityError, 'fixture rejects insert'): + self.record() + self.assertEqual(self.snapshot(), before) + + def test_failed_new_event_rolls_back_reopened_row_and_history(self): + self.record() + self.db.mark_notified('security_updates') + self.db.resolve_error('security_updates') + before = self.snapshot() + old_id = before[0][0]['id'] + self.assertEqual([event['event_type'] for event in before[1]], ['new', 'resolved']) + conn = sqlite3.connect(str(self.db.db_path)) + try: + # Fail at the later durable-event insert, only after the old row + # has been replaced by the fresh episode within this transaction. + conn.execute(f'''CREATE TRIGGER fixture_reject_new_event BEFORE INSERT ON events + WHEN NEW.event_type = 'new' AND NEW.error_key = 'security_updates' + BEGIN + SELECT RAISE(ABORT, 'fixture rejects new event') + WHERE EXISTS ( + SELECT 1 FROM errors + WHERE error_key = NEW.error_key AND id <> {old_id} + AND resolved_at IS NULL AND first_seen = NEW.timestamp + ); + END''') + conn.commit() + finally: + conn.close() + self.advance(minutes=1) + with self.assertRaisesRegex(sqlite3.IntegrityError, 'fixture rejects new event'): + self.record() + self.assertEqual(self.snapshot(), before) + + +if __name__ == '__main__': + unittest.main() diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index f5b7587c..aab13a74 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -863,8 +863,11 @@ publication workflow and has read-only repository permissions. #### Qualified offline `tests/` CI lanes `.github/workflows/test-offline-qualified.yml` runs **only** the explicitly listed -eight Python files (including the inert active-notification payload fixture) and eleven +nine Python files (including the inert active-notification payload fixture and +the focused health recurrence tests) and eleven Node files in `.github/scripts/run_offline_qualified.py`. +The Python runner gives each test file a private temporary directory through +`TMPDIR`. It removes that directory after the test file finishes. Run the same allowlist locally from the root of a clean, disposable checkout with no `AppImage/node_modules`. Do not run this over a normal Monitor install: the commands refuse an existing directory, file, or symlink (including a @@ -892,7 +895,7 @@ React peer constraint against locked React 19; it does not resolve that mismatch tests, skipped Python tests, or a nonzero exit from either lane. Node fixtures use top-level assertions, not a runner that reports case/skip totals. These are local fixture tests, not the whole `tests/` tree, `AppImage/tests/`, -`AppImage/scripts/tests/`, browser smoke, a live Proxmox host, or an AppImage +the whole `AppImage/scripts/tests/` tree, browser smoke, a live Proxmox host, or an AppImage build. The CI workflow has read-only permissions and runs on relevant PRs and main/develop pushes, or manually. Review the input paths and the test's imports before extending the explicit manifest.