50 lines
2.8 KiB
Python
50 lines
2.8 KiB
Python
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 faults
|
|
|
|
|
|
class FaultTests(unittest.TestCase):
|
|
def test_repeat_exit_and_stable_recovery_report_once(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
client=SimpleNamespace(call=lambda *args:None)
|
|
records={}
|
|
with patch.object(faults,'Client',return_value=client),patch.object(client,'call') as call,patch.object(faults.time,'time',return_value=100):
|
|
for _ in range(3):faults.report_process({'control':{}},Path(root),'mirror',SimpleNamespace(poll=lambda:2),records)
|
|
self.assertEqual(call.call_count,1)
|
|
incident=records['mirror']['event_id']
|
|
(Path(root)/'mirror-health.json').write_text('{"status":"ok","updated":100}')
|
|
faults.report_process({'control':{}},Path(root),'mirror',SimpleNamespace(poll=lambda:None),records)
|
|
with patch.object(faults,'Client',return_value=client),patch.object(client,'call') as call,patch.object(faults.time,'time',return_value=131):
|
|
faults.report_process({'control':{}},Path(root),'mirror',SimpleNamespace(poll=lambda:None),records)
|
|
self.assertEqual(call.call_args.args[1]['event_id'],incident)
|
|
self.assertEqual(call.call_args.args[1]['status'],'resolved')
|
|
self.assertEqual(records,{})
|
|
|
|
def test_recovery_delivery_failure_then_exit_never_sends_recovery(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
c=SimpleNamespace(call=lambda *args:None)
|
|
records={'mirror':{'event_id':'incident','state':'firing','pending':'resolved','delivery':'failed'}}
|
|
with patch.object(faults,'Client',return_value=c),patch.object(c,'call') as call:
|
|
faults.report_process({'control':{}},Path(root),'mirror',SimpleNamespace(poll=lambda:2),records)
|
|
call.assert_not_called()
|
|
self.assertTrue(records['mirror']['needs_reconciliation'])
|
|
self.assertEqual(records['mirror']['delivery'],'uncertain_recovery')
|
|
|
|
def test_hook_failure_remains_persistently_visible(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
c=SimpleNamespace(call=lambda *args:None)
|
|
records={}
|
|
with patch.object(faults,'Client',return_value=c),patch.object(c,'call',side_effect=RuntimeError('offline')):
|
|
faults.report_process({'control':{}},Path(root),'mirror',SimpleNamespace(poll=lambda:2),records)
|
|
self.assertEqual(records['mirror']['delivery'],'failed')
|
|
self.assertEqual(records['mirror']['pending'],'firing')
|
|
self.assertTrue((Path(root)/'bridge-faults.json').exists())
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|