76 lines
4.0 KiB
Python
76 lines
4.0 KiB
Python
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
|
|
import notifier
|
|
|
|
|
|
class NotifierTests(unittest.TestCase):
|
|
def fixture(self,root):
|
|
cfg={'control':{},'notification_targets':{'control':'11111111-1111-4111-8111-111111111111'},'codex':'codex','cwd':root}
|
|
n={'notification_id':'note1','target':'control','hook':{'source':'test','event_id':'incident','summary':'synthetic failure','severity':'warning','status':'firing'}}
|
|
return cfg,n,Path(root)/'state.json'
|
|
|
|
def test_ack_retry_never_resubmits_accepted_queue(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg,n,path=self.fixture(root)
|
|
class Client:
|
|
failed=False
|
|
def call(self,p,data=None):
|
|
if p.endswith('notifications'):return {'notifications':[n]}
|
|
if not self.failed:self.failed=True;raise RuntimeError('ack offline')
|
|
return {}
|
|
c=Client();state={}
|
|
result=SimpleNamespace(returncode=0,stdout='Queued message 22222222-2222-4222-8222-222222222222 for thread '+cfg['notification_targets']['control']+'.')
|
|
with patch.object(notifier.subprocess,'run',return_value=result) as run:
|
|
with self.assertRaises(RuntimeError):notifier.deliver(c,cfg,state,path)
|
|
notifier.deliver(c,cfg,state,path)
|
|
self.assertEqual(run.call_count,1)
|
|
self.assertTrue(state['deliveries']['note1']['acked'])
|
|
|
|
def test_uncertain_queue_never_blindly_replays(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg,n,path=self.fixture(root)
|
|
class Client:
|
|
def call(self,p,data=None):return {'notifications':[n]}
|
|
class Reader:
|
|
def __init__(self,*args):pass
|
|
def __enter__(self):return self
|
|
def __exit__(self,*args):pass
|
|
def history(self,thread):return {'id':thread},[]
|
|
state={}
|
|
with patch.object(notifier.subprocess,'run',return_value=SimpleNamespace(returncode=1,stdout='')) as run,patch.object(notifier,'AppServer',Reader):
|
|
with self.assertRaisesRegex(RuntimeError,'uncertain'):notifier.deliver(Client(),cfg,state,path)
|
|
with self.assertRaisesRegex(RuntimeError,'reconciliation'):notifier.deliver(Client(),cfg,state,path)
|
|
self.assertEqual(run.call_count,1)
|
|
|
|
def test_unknown_target_does_not_block_other_target(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg,n,path=self.fixture(root)
|
|
cfg['notification_targets']['human']='33333333-3333-4333-8333-333333333333'
|
|
other={**n,'notification_id':'note2','target':'human'}
|
|
class Client:
|
|
def call(self,p,data=None):return {'notifications':[n,other]} if p.endswith('notifications') else {}
|
|
results=[SimpleNamespace(returncode=1,stdout=''),SimpleNamespace(returncode=0,stdout='Queued message 22222222-2222-4222-8222-222222222222 for thread '+cfg['notification_targets']['human']+'.')]
|
|
state={}
|
|
with patch.object(notifier.subprocess,'run',side_effect=results) as run:
|
|
with self.assertRaises(RuntimeError):notifier.deliver(Client(),cfg,state,path)
|
|
self.assertEqual(run.call_count,2)
|
|
self.assertTrue(state['deliveries']['note2']['acked'])
|
|
self.assertEqual(state['deliveries']['note1']['state'],'submitting')
|
|
|
|
def test_no_events_means_no_conversation_input(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg,n,path=self.fixture(root)
|
|
c=SimpleNamespace(call=lambda *args:{'notifications':[]})
|
|
with patch.object(notifier.subprocess,'run') as run:
|
|
notifier.deliver(c,cfg,{},path)
|
|
run.assert_not_called()
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|