74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
import os, sys, types, unittest
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload', 'user', 'remote_access', 'pager-webui'))
|
|
import mk8_gate
|
|
|
|
|
|
class DecisionTest(unittest.TestCase):
|
|
def tick(self, state):
|
|
action, new = mk8_gate.watchdog_decision(state)
|
|
return action, new
|
|
|
|
def test_no_action_below_fail_threshold(self):
|
|
action, s = self.tick({'fails': 5, 'oks': 0, 'tripped': False})
|
|
self.assertIsNone(action)
|
|
self.assertFalse(s['tripped'])
|
|
|
|
def test_rollback_at_threshold(self):
|
|
action, s = self.tick({'fails': 6, 'oks': 0, 'tripped': False})
|
|
self.assertEqual(action, 'rollback')
|
|
self.assertTrue(s['tripped'])
|
|
self.assertEqual(s['oks'], 0)
|
|
|
|
def test_promote_after_recovery(self):
|
|
action, s = self.tick({'fails': 6, 'oks': 6, 'tripped': True})
|
|
self.assertEqual(action, 'promote')
|
|
|
|
def test_no_promote_before_recovery_threshold(self):
|
|
action, s = self.tick({'fails': 6, 'oks': 5, 'tripped': True})
|
|
self.assertIsNone(action)
|
|
|
|
|
|
class EnterTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.addCleanup(setattr, mk8_gate, 'ENABLED', mk8_gate.ENABLED)
|
|
self.spawned = []
|
|
|
|
def _install_fake_profiles(self):
|
|
snaps = []
|
|
fake = types.ModuleType('mk8_profiles')
|
|
fake.auto_name = lambda op: 'pre-%s-42' % op
|
|
fake.snapshot = lambda name: (snaps.append(name), True)[1]
|
|
old = sys.modules.get('mk8_profiles')
|
|
sys.modules['mk8_profiles'] = fake
|
|
self.addCleanup(sys.modules.__setitem__, 'mk8_profiles', old)
|
|
return snaps
|
|
|
|
def _capture_popen(self):
|
|
cmds = []
|
|
old = mk8_gate.subprocess.Popen
|
|
mk8_gate.subprocess.Popen = lambda cmd, **kw: cmds.append(cmd)
|
|
self.addCleanup(setattr, mk8_gate.subprocess, 'Popen', old)
|
|
return cmds
|
|
|
|
def test_enter_disabled_is_noop(self):
|
|
mk8_gate.ENABLED = False
|
|
cmds = self._capture_popen()
|
|
snaps = self._install_fake_profiles()
|
|
self.assertIsNone(mk8_gate.enter('ap_change'))
|
|
self.assertEqual(snaps, [])
|
|
self.assertEqual(cmds, [])
|
|
|
|
def test_enter_enabled_snapshots_and_spawns(self):
|
|
mk8_gate.ENABLED = True
|
|
cmds = self._capture_popen()
|
|
snaps = self._install_fake_profiles()
|
|
name = mk8_gate.enter('attack_deploy')
|
|
self.assertEqual(name, 'pre-attack_deploy-42')
|
|
self.assertEqual(snaps, [name])
|
|
self.assertEqual(len(cmds), 1)
|
|
self.assertIn('setsid sh', cmds[0])
|
|
self.assertIn(name, cmds[0])
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|