esh-pfi-infrastructure/services/althing-alert-bridge/test_bridge.py

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()