import importlib.util import os from pathlib import Path import subprocess import unittest from unittest.mock import patch spec=importlib.util.spec_from_file_location('bridge',Path(__file__).with_name('bridge.py')) bridge=importlib.util.module_from_spec(spec);spec.loader.exec_module(bridge) class DeliveryTests(unittest.TestCase): def setUp(self): self.env=patch.dict(os.environ,POSTBOX='/bin/postbox',BESZEL_ALERT_RECIPIENT='infra-ops') self.env.start();self.addCleanup(self.env.stop) def test_posts_body_as_stdin_and_returns_receipt(self): with patch.object(bridge.subprocess,'run',return_value=subprocess.CompletedProcess([],0,'{"message_id": 42}','')) as run: self.assertEqual(bridge.deliver({'title':'Disk 85%','message':'tank is full'})['message_id'],42) args,kw=run.call_args self.assertEqual(args[0],['/bin/postbox','--json','send','--to','infra-ops','--subject','[Beszel] Disk 85%']) self.assertIn('tank is full',kw['input']) def test_delivery_failure_is_not_success(self): with patch.object(bridge.subprocess,'run',return_value=subprocess.CompletedProcess([],2,'','post office unavailable')): with self.assertRaises(RuntimeError):bridge.deliver({'message':'disk full'}) def test_rejects_empty_message_without_sending(self): with patch.object(bridge.subprocess,'run') as run: with self.assertRaises(ValueError):bridge.deliver({'message':''}) run.assert_not_called() class KumaRouteTests(unittest.TestCase): """The /kuma route. The point of generalising was that an alert must not lie about its own source -- a Kuma outage labelled [Beszel] sends you to the wrong dashboard, which is worse than no alert.""" def setUp(self): self.env=patch.dict(os.environ,POSTBOX='/bin/postbox',ALERT_RECIPIENT='infra-ops') self.env.start();self.addCleanup(self.env.stop) DOWN={'monitor':{'name':'Homepage','url':'http://10.0.50.45:5100/'}, 'heartbeat':{'status':0,'msg':'connect ECONNREFUSED'}, 'msg':'[Homepage] [Down] connect ECONNREFUSED'} def test_subject_names_the_service_and_the_direction(self): with patch.object(bridge.subprocess,'run',return_value=subprocess.CompletedProcess([],0,'{"message_id":7}','')) as run: bridge.deliver(self.DOWN,'/kuma') argv=run.call_args[0][0] self.assertIn('[Uptime Kuma] Homepage is DOWN',argv) self.assertNotIn('[Beszel] Homepage is DOWN',argv) def test_up_transition_reads_as_up(self): up={'monitor':{'name':'Gitea'},'heartbeat':{'status':1,'msg':'200 - OK'},'msg':'[Gitea] [Up] 200 - OK'} with patch.object(bridge.subprocess,'run',return_value=subprocess.CompletedProcess([],0,'{"message_id":8}','')) as run: bridge.deliver(up,'/kuma') self.assertIn('[Uptime Kuma] Gitea is UP',run.call_args[0][0]) def test_body_carries_the_target_and_the_kuma_board_not_the_beszel_hub(self): with patch.object(bridge.subprocess,'run',return_value=subprocess.CompletedProcess([],0,'{"message_id":9}','')) as run: bridge.deliver(self.DOWN,'/kuma') body=run.call_args[1]['input'] self.assertIn('http://10.0.50.45:5100/',body) self.assertIn('Board: http://10.250.50.70:3001',body) self.assertNotIn('Hub: http://10.250.50.70:8090',body) # the Beszel footer must not leak def test_beszel_footer_is_unchanged_by_the_refactor(self): with patch.object(bridge.subprocess,'run',return_value=subprocess.CompletedProcess([],0,'{"message_id":10}','')) as run: bridge.deliver({'title':'Disk 85%','message':'tank is full'},'/beszel') body=run.call_args[1]['input'] self.assertIn('Hub: http://10.250.50.70:8090',body) self.assertNotIn('Board:',body) def test_a_nameless_alert_prefers_its_message_over_a_fabricated_subject(self): """Superseded an earlier assertion that this should read "unknown monitor is DOWN". It should not: a real monitor always carries a name, so a nameless payload is degenerate, and "something broke" tells the reader more than a placeholder that looks like a bug.""" with patch.object(bridge.subprocess,'run',return_value=subprocess.CompletedProcess([],0,'{"message_id":11}','')) as run: argv=None bridge.deliver({'heartbeat':{'status':0},'msg':'something broke'},'/kuma') argv=run.call_args[0][0] self.assertIn('[Uptime Kuma] something broke',argv) self.assertNotIn('unknown monitor',' '.join(argv)) def test_empty_kuma_message_is_refused_without_sending(self): with patch.object(bridge.subprocess,'run') as run: with self.assertRaises(ValueError): bridge.deliver({'monitor':{'name':'X'},'heartbeat':{'status':0},'msg':''},'/kuma') run.assert_not_called() def test_a_monitorless_notification_uses_its_own_message_as_the_subject(self): """testNotification and cert-expiry arrive with no monitor and no heartbeat. Fabricating "unknown monitor is ?" from those makes a real alert read like a bug -- observed live on 2026-09-21.""" with patch.object(bridge.subprocess,'run',return_value=subprocess.CompletedProcess([],0,'{"message_id":13}','')) as run: bridge.deliver({'msg':'althing (infra-ops) Testing'},'/kuma') argv=run.call_args[0][0] self.assertIn('[Uptime Kuma] althing (infra-ops) Testing',argv) self.assertNotIn('unknown monitor',' '.join(argv)) self.assertNotIn('is ?',' '.join(argv)) def test_legacy_beszel_env_names_still_resolve(self): """A half-finished deploy (new bridge, old unit) must start, not crash.""" with patch.dict(os.environ,{'ALERT_RECIPIENT':'','BESZEL_ALERT_RECIPIENT':'infra-ops'}): with patch.object(bridge.subprocess,'run',return_value=subprocess.CompletedProcess([],0,'{"message_id":12}','')) as run: bridge.deliver({'title':'t','message':'m'}) self.assertIn('infra-ops',run.call_args[0][0]) if __name__ == '__main__': unittest.main()