mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-10-09 06:56:37 +00:00
160 lines
10 KiB
Python
160 lines
10 KiB
Python
"""Maintainer acceptance at inert actual notification consumers."""
|
|
import unittest
|
|
from notification_fixture import templates, receive, LANGUAGES, SCRIPTS, extract
|
|
from notification_final_fixture import deliver
|
|
|
|
NATIVE_REPORT = '''Details
|
|
=======
|
|
VMID Name Status Time Size Filename
|
|
100 web ok 1m 1s 1 GiB vm/100/2026-09-29T17:00:00Z
|
|
|
|
Total running time: 1m 1s
|
|
Total size: 1 GiB
|
|
|
|
Logs
|
|
====
|
|
vzdump --all 1 --storage PBS --mode snapshot
|
|
|
|
100: 2026-09-29 17:00:00 INFO: Starting Backup of VM 100 (qemu)
|
|
100: 2026-09-29 17:01:01 INFO: Finished Backup of VM 100 (00:01:01)
|
|
'''
|
|
|
|
class MaintainerFollowupTests(unittest.TestCase):
|
|
def test_completed_with_warnings_requires_independent_completion(self):
|
|
warning = '\n100: 2026-09-29 17:00:01 WARN: file changed during backup'
|
|
event = receive(NATIVE_REPORT + warning)
|
|
self.assertEqual(event.data['backup_outcome'], 'completed_with_warnings')
|
|
self.assertEqual((event.event_type,event.severity), ('backup_complete','INFO'))
|
|
for manual in (False,True):
|
|
for lang in LANGUAGES:
|
|
result = deliver(event.event_type,event.data,event.severity,lang,manual=manual)
|
|
label = templates.runtime_message('backup.warningTitle',lang,hostname='node-a')
|
|
self.assertIn(label,result['title'])
|
|
status = templates.runtime_message('channels.email.status.completed_with_warnings',lang)
|
|
self.assertIn(status,result['text'])
|
|
self.assertIn('file changed during backup',result['text'])
|
|
# Manual sends intentionally skip channel emoji enrichment.
|
|
if not manual: self.assertTrue(result['title'].startswith('💾⚠️'))
|
|
rich, _ = templates.enrich_with_emojis(event.event_type,result['title'],result['body'],event.data)
|
|
self.assertTrue(rich.startswith('💾⚠️'))
|
|
self.assertEqual(receive('WARN: file changed during backup').data['backup_outcome'],'unconfirmed')
|
|
self.assertEqual(receive(NATIVE_REPORT.split('Total running time:')[0]+warning).data['backup_outcome'],'unconfirmed')
|
|
self.assertEqual(receive(NATIVE_REPORT+warning+'\nERROR: cleanup failed').data['backup_outcome'],'failed')
|
|
self.assertEqual(receive(NATIVE_REPORT+warning,'warning').data['backup_outcome'],'completed_with_warnings')
|
|
self.assertEqual(receive('INFO: Starting Backup of VM 100 (qemu)\nINFO: Finished Backup of VM 100 (00:01:01)'+warning).data['backup_outcome'],'completed_with_warnings')
|
|
self.assertEqual(receive(NATIVE_REPORT+warning,kind='').data['backup_outcome'],'unconfirmed')
|
|
|
|
def test_null_filename_failed_guest_uses_own_start_identity(self):
|
|
report = NATIVE_REPORT.replace('100 web ok 1m 1s 1 GiB vm/100/2026-09-29T17:00:00Z',
|
|
'100 web err 1m 1s 0 B null')
|
|
for kind, prefix in (('qemu','VM'),('lxc','CT')):
|
|
own = report.replace('VM 100 (qemu)',f'VM 100 ({kind})')
|
|
# Unrelated guest appears first and must never supply the failed type.
|
|
message = f'INFO: Starting Backup of VM 999 ({"lxc" if kind == "qemu" else "qemu"})\n' + own
|
|
event = receive(message,'error','vzdump backup status (raw-node): backup failed')
|
|
parsed = templates._parse_vzdump_message(message)
|
|
self.assertEqual(parsed['vms'][0]['type'],kind)
|
|
result = deliver(event.event_type,event.data,event.severity)
|
|
self.assertIn(f'{prefix} web (100)',result['title'])
|
|
self.assertIn(f'❌ {prefix} web (100)',result['body'])
|
|
for message in (report.replace('Starting Backup of VM 100','Starting Backup of VM 999'),
|
|
report+'\nINFO: Starting Backup of VM 100 (lxc)'):
|
|
self.assertEqual(templates._parse_vzdump_message(message)['vms'][0]['type'],'')
|
|
|
|
def test_native_subject_keeps_cause_without_host_envelope(self):
|
|
subject = 'vzdump backup status (raw-host): backup failed: multiple problems'
|
|
event = receive('ERROR: archive write failed\n'+NATIVE_REPORT,'error',subject)
|
|
event.data['hostname']='configured-alias'
|
|
for lang in LANGUAGES:
|
|
result=deliver(event.event_type,event.data,event.severity,lang)
|
|
self.assertNotIn(subject,result['text'])
|
|
self.assertEqual(result['text'].count('ERROR: archive write failed'),1)
|
|
self.assertIn('configured-alias',result['title'])
|
|
setup=receive('Details\n=======\nVMID Name Status Time Size Filename\n\nTotal running time: 0s\nTotal size: 0 B','error',subject.replace('multiple problems','unable to open storage'))
|
|
result=deliver(setup.event_type,setup.data,setup.severity)
|
|
self.assertEqual(result['text'].count('unable to open storage'),1)
|
|
self.assertNotIn(setup.data['pve_title'],result['text'])
|
|
unique=receive(NATIVE_REPORT,'error',subject.replace('multiple problems','job-end hook denied'))
|
|
result=deliver(unique.event_type,unique.data,unique.severity)
|
|
self.assertIn('job-end hook denied',result['body'])
|
|
self.assertNotIn('vzdump backup status',result['body'])
|
|
|
|
def test_backup_diagnostics_are_bounded_with_principal_cause_and_notice(self):
|
|
raw = NATIVE_REPORT + '\n' + '\n'.join(f'WARN: repeated warning {i}' for i in range(80)) + '\nERROR: principal archive write failure'
|
|
event=receive(raw)
|
|
for lang in LANGUAGES:
|
|
result=deliver(event.event_type,event.data,event.severity,lang)
|
|
self.assertIn('ERROR: principal archive write failure',result['body'])
|
|
self.assertLessEqual(len('\n'.join(line for line in result['body'].splitlines() if line.startswith(('WARN:','ERROR:')))),1024)
|
|
self.assertLessEqual(sum(line.startswith(('WARN:','ERROR:')) for line in result['body'].splitlines()),8)
|
|
notice=templates.runtime_message('backup.diagnosticsOmitted',lang,count=73)
|
|
self.assertIn(notice,result['body'])
|
|
self.assertEqual(event.data['pve_message'],raw)
|
|
long=receive('ERROR: '+ 'b'*5000,'error','vzdump backup status (node): backup failed')
|
|
result=deliver(long.event_type,long.data,long.severity)
|
|
self.assertLess(len(result['body']),1400)
|
|
self.assertIn('ERROR: '+ 'b'*100,result['body'])
|
|
self.assertIn(templates.runtime_message('backup.diagnosticsOmitted','en',count=1),result['body'])
|
|
|
|
def test_real_quiet_digest_retains_each_backup_outcome_icon(self):
|
|
samples=[(NATIVE_REPORT,'confirmed','💾✅'),(NATIVE_REPORT+'\nWARN: changed file','completed_with_warnings','💾⚠️'),(NATIVE_REPORT+'\nERROR: write failed','failed','💾❌'),('INFO: Starting Backup of VM 100 (qemu)','unconfirmed','💾❔')]
|
|
for raw,outcome,icon in samples:
|
|
event=receive(raw)
|
|
self.assertEqual(event.data['backup_outcome'],outcome)
|
|
for lang in LANGUAGES:
|
|
result=deliver(event.event_type,event.data,event.severity,lang,quiet=True)
|
|
self.assertIn(icon,result['body'])
|
|
self.assertNotIn('💾❔',result['body']) if outcome!='unconfirmed' else None
|
|
import types,datetime
|
|
ns={'datetime':datetime.datetime,'runtime_message':templates.runtime_message,'EVENT_EMOJI':templates.EVENT_EMOJI,'CATEGORY_EMOJI':templates.CATEGORY_EMOJI}
|
|
compose=extract(SCRIPTS/'notification_manager.py','_compose_digest_body','NotificationManager',ns)
|
|
target=types.SimpleNamespace(_notification_language=lambda:'en')
|
|
rows=[(i,'backup_complete','backup',1,icon+' node: Backup','') for i,(_,_,icon) in enumerate(samples)]
|
|
body=compose(target,rows,use_icons=True)
|
|
for _,_,icon in samples:self.assertIn(icon,body)
|
|
plain=compose(target,rows,use_icons=False)
|
|
for _,_,icon in samples:self.assertNotIn(icon,plain)
|
|
should=extract(SCRIPTS/'notification_manager.py','_should_buffer_for_digest','NotificationManager',{})
|
|
self.assertFalse(should(types.SimpleNamespace(_DIGEST_EXEMPT_EVENTS={'backup_complete'},_config={'email.digest_enabled':'true'}),'email','INFO','backup_complete'))
|
|
|
|
def test_quiet_restore_details_are_subordinate_in_text_and_email(self):
|
|
from notification_final_fixture import restore_event
|
|
event=restore_event('missing module zfs')
|
|
for lang in LANGUAGES:
|
|
result=deliver(event['event_type'],event['data'],event['severity'],lang,quiet=True)
|
|
body_lines=result['buffered'][0][3].splitlines()
|
|
for line in body_lines:
|
|
if line.strip():
|
|
self.assertIn(' '+line.strip(),result['body'])
|
|
self.assertIn(' '+__import__('html').escape(line.strip()),result['html'])
|
|
self.assertIn('white-space:pre-wrap;',result['html'])
|
|
self.assertNotIn(templates.runtime_message('digest.footer',lang),result['body'])
|
|
|
|
def test_concrete_subject_cause_already_in_error_log_is_not_repeated(self):
|
|
event=receive(NATIVE_REPORT+'\n100: ERROR: job-end hook denied','error',
|
|
'vzdump backup status (raw-host): backup failed: job-end hook denied')
|
|
for lang in LANGUAGES:
|
|
result=deliver(event.event_type,event.data,event.severity,lang)
|
|
self.assertEqual(result['text'].count('job-end hook denied'),1)
|
|
self.assertIn('100: ERROR: job-end hook denied',result['body'])
|
|
|
|
def test_job_level_subject_cause_survives_warning_cap(self):
|
|
raw=NATIVE_REPORT+'\n'+'\n'.join('WARN: repeated diagnostic '+str(i) for i in range(80))
|
|
event=receive(raw,'error','vzdump backup status (raw-host): backup failed: job-end hook denied')
|
|
for lang in LANGUAGES:
|
|
result=deliver(event.event_type,event.data,event.severity,lang)
|
|
self.assertEqual(result['body'].count('job-end hook denied'),1)
|
|
self.assertNotIn('vzdump backup status',result['body'])
|
|
self.assertIn(templates.runtime_message('backup.diagnosticsOmitted',lang,count=73),result['body'])
|
|
|
|
def test_backup_quiet_email_uses_event_scoped_wrapping(self):
|
|
event=receive(NATIVE_REPORT+'\nWARN: changed file')
|
|
result=deliver(event.event_type,event.data,event.severity,'sv',quiet=True)
|
|
self.assertTrue(result['data'].get('_backup_summary'))
|
|
self.assertIn('table-layout:fixed;',result['html'])
|
|
self.assertIn('overflow-wrap:break-word;',result['html'])
|
|
unrelated=deliver('node_reconnect',{'hostname':'node-a'},'OK')
|
|
self.assertNotIn('table-layout:fixed;',unrelated['html'])
|
|
|
|
if __name__ == '__main__': unittest.main()
|