Wire Beszel fleet filesystems, GPU telemetry, dashboard and alerts

services/beszel-althing/test_bridge.py

@@ -0,0 +1,30 @@
+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()
+
+if __name__ == '__main__':
+    unittest.main()