feat: verstuur dashcam-notificatie via Matrix (sherpa) i.p.v. WhatsApp
wwebjs-api is vervangen door wuzapi; notificatiekanaal is nu Matrix: - send_matrix_notification() POST -> sherpa compat-token uit ~/whatsed-new/secrets/sherpa-token naar eddy-report-room - zelfde patronen als watchdog-report/WAN IP Guard - tests: SendWhatsappTest -> SendMatrixTest (+2 asserts)
This commit is contained in:
parent
0227dbda3f
commit
55c282906e
3 changed files with 88 additions and 40 deletions
|
|
@ -215,36 +215,67 @@ class EnsureFreeSpaceTest(unittest.TestCase):
|
|||
dashcam.ensure_free_space(10_000, self.base)
|
||||
|
||||
|
||||
class SendWhatsappTest(unittest.TestCase):
|
||||
class SendMatrixTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
patcher = mock.patch.object(dashcam.urllib.request, "urlopen")
|
||||
self.mock_urlopen = patcher.start()
|
||||
self.mock_urlopen.return_value.__enter__.return_value = mock.Mock()
|
||||
self.addCleanup(patcher.stop)
|
||||
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
token_file = os.path.join(self.tmp.name, "sherpa-token")
|
||||
room_file = os.path.join(self.tmp.name, "eddy-report-room")
|
||||
with open(token_file, "w") as f:
|
||||
f.write("mct_testtoken\n")
|
||||
with open(room_file, "w") as f:
|
||||
f.write("!testroom:whatsed.eddydevink.nl\n")
|
||||
p1 = mock.patch.object(dashcam, "MATRIX_TOKEN_FILE", token_file)
|
||||
p2 = mock.patch.object(dashcam, "MATRIX_ROOM_FILE", room_file)
|
||||
p1.start()
|
||||
p2.start()
|
||||
self.addCleanup(p1.stop)
|
||||
self.addCleanup(p2.stop)
|
||||
|
||||
def _body(self):
|
||||
req = self.mock_urlopen.call_args[0][0]
|
||||
return json.loads(req.data)["content"]
|
||||
return json.loads(req.data)["body"]
|
||||
|
||||
def _request(self):
|
||||
return self.mock_urlopen.call_args[0][0]
|
||||
|
||||
def test_success_message(self):
|
||||
dashcam.send_whatsapp_notification(5, 0, 381_000_000)
|
||||
dashcam.send_matrix_notification(5, 0, 381_000_000)
|
||||
self.assertIn("5 bestanden", self._body())
|
||||
self.assertIn("geen fouten", self._body())
|
||||
|
||||
def test_errors_message(self):
|
||||
dashcam.send_whatsapp_notification(3, 2, 100_000_000)
|
||||
dashcam.send_matrix_notification(3, 2, 100_000_000)
|
||||
self.assertIn("2 fout(en)", self._body())
|
||||
|
||||
def test_aborted_message(self):
|
||||
dashcam.send_whatsapp_notification(
|
||||
dashcam.send_matrix_notification(
|
||||
1, 4, 100_000_000, aborted=True, reason="SD-kaart onleesbaar"
|
||||
)
|
||||
self.assertIn("GESTOPT", self._body())
|
||||
self.assertIn("SD-kaart onleesbaar", self._body())
|
||||
|
||||
def test_sends_room_message_with_bearer_token(self):
|
||||
dashcam.send_matrix_notification(1, 0, 10_000)
|
||||
req = self._request()
|
||||
self.assertEqual(req.get_method(), "PUT")
|
||||
self.assertEqual(req.get_header("Authorization"), "Bearer mct_testtoken")
|
||||
self.assertIn("/rooms/!testroom:whatsed.eddydevink.nl/send/m.room.message/", req.full_url)
|
||||
self.assertEqual(json.loads(req.data)["msgtype"], "m.text")
|
||||
|
||||
def test_missing_secrets_is_swallowed(self):
|
||||
with mock.patch.object(dashcam, "MATRIX_TOKEN_FILE", "/nonexistent/token"):
|
||||
dashcam.send_matrix_notification(0, 0, 0)
|
||||
self.mock_urlopen.assert_not_called()
|
||||
|
||||
def test_api_failure_is_swallowed(self):
|
||||
self.mock_urlopen.side_effect = Exception("boom")
|
||||
dashcam.send_whatsapp_notification(0, 0, 0)
|
||||
dashcam.send_matrix_notification(0, 0, 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
|||
Loading…
Reference in a new issue