118 lines
4.4 KiB
Python
118 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Pins plugcut.py: its refusals, the countdown sent WITH the off, the watchdog armed BEFORE the off,
|
|
the fail-safe when the countdown does not show, and the `finally:` re-arm. A fake plug, no network."""
|
|
import argparse, os, sys, unittest
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import plugcut # noqa: E402
|
|
|
|
|
|
def args(**kw):
|
|
d = dict(plug='hp', off_seconds=10, i_mean_a_power_cut=True, long=False, allow_n100=False)
|
|
d.update(kw)
|
|
return argparse.Namespace(**d)
|
|
|
|
|
|
class FakePlug:
|
|
"""Records every command. `countdown_works`: the plug turns itself on when the countdown ends."""
|
|
def __init__(self, countdown_works=True, show_countdown=True, fail_after_off=False):
|
|
self.on, self.cd, self.log = True, 0, []
|
|
self.countdown_works, self.show_countdown, self.fail_after_off = countdown_works, show_countdown, fail_after_off
|
|
self.clock = 0.0
|
|
|
|
def set_multiple_values(self, v):
|
|
self.log.append(('set', dict(v)))
|
|
if '1' in v:
|
|
self.on = v['1']
|
|
if '9' in v and self.show_countdown:
|
|
self.cd = v['9']
|
|
|
|
def turn_on(self):
|
|
self.log.append(('on',))
|
|
self.on, self.cd = True, 0
|
|
|
|
def status(self):
|
|
if self.fail_after_off and not self.on and any(e[0] == 'status' for e in self.log):
|
|
self.log.append(('status',))
|
|
raise RuntimeError('plug gone')
|
|
self.log.append(('status',))
|
|
return {'dps': {'1': self.on, '9': self.cd}}
|
|
|
|
def tick(self, s):
|
|
self.clock += s
|
|
if self.countdown_works and not self.on and self.cd and self.clock >= self.cd:
|
|
self.on, self.cd = True, 0
|
|
|
|
|
|
class Refusals(unittest.TestCase):
|
|
def test_allowed(self):
|
|
self.assertIsNone(plugcut.refuse_reason(args()))
|
|
|
|
def test_without_the_long_flag(self):
|
|
self.assertIn('--i-mean-a-power-cut', plugcut.refuse_reason(args(i_mean_a_power_cut=False)))
|
|
|
|
def test_n100_without_operator_word(self):
|
|
self.assertIn('N100', plugcut.refuse_reason(args(plug='n100')))
|
|
self.assertIsNone(plugcut.refuse_reason(args(plug='n100', allow_n100=True)))
|
|
|
|
def test_unknown_plug(self):
|
|
self.assertIn('unknown', plugcut.refuse_reason(args(plug='dooplex')))
|
|
|
|
def test_over_60_needs_long(self):
|
|
self.assertIn('--long', plugcut.refuse_reason(args(off_seconds=61)))
|
|
self.assertIsNone(plugcut.refuse_reason(args(off_seconds=3000, long=True)))
|
|
|
|
def test_long_capped_at_3600(self):
|
|
self.assertIsNotNone(plugcut.refuse_reason(args(off_seconds=3601, long=True)))
|
|
|
|
def test_zero_seconds(self):
|
|
self.assertIsNotNone(plugcut.refuse_reason(args(off_seconds=0)))
|
|
|
|
|
|
class Cut(unittest.TestCase):
|
|
def run_cut(self, plug, off_s=10):
|
|
order = []
|
|
def watchdog(p, after):
|
|
order.append(('watchdog', after, len(plug.log)))
|
|
return 4242
|
|
rc = plugcut.cut(plug, 'hp', off_s, watchdog=watchdog, sleep=plug.tick, now=lambda: plug.clock)
|
|
return rc, order
|
|
|
|
def test_countdown_sent_with_the_off(self):
|
|
p = FakePlug()
|
|
rc, _ = self.run_cut(p)
|
|
first_off = next(e for e in p.log if e[0] == 'set' and e[1].get('1') is False)
|
|
self.assertEqual(first_off[1].get('9'), 10, 'the off must carry the countdown in the same command')
|
|
self.assertEqual(rc, 0)
|
|
self.assertTrue(p.on)
|
|
|
|
def test_watchdog_armed_before_the_off(self):
|
|
p = FakePlug()
|
|
_, order = self.run_cut(p)
|
|
self.assertEqual(order[0][2], 0, 'the watchdog must start before any command reaches the plug')
|
|
self.assertEqual(order[0][1], 10 + plugcut.WATCHDOG_EXTRA_S)
|
|
|
|
def test_failsafe_when_countdown_not_shown(self):
|
|
p = FakePlug(show_countdown=False)
|
|
rc, _ = self.run_cut(p)
|
|
self.assertTrue(p.on)
|
|
self.assertIn(('on',), p.log)
|
|
self.assertEqual(rc, 0)
|
|
|
|
def test_finally_rearms_when_countdown_never_fires(self):
|
|
p = FakePlug(countdown_works=False)
|
|
rc, _ = self.run_cut(p)
|
|
self.assertEqual(rc, 1, 'a countdown that did not fire is a failed cut')
|
|
self.assertTrue(p.on, 'finally: must switch the relay back on')
|
|
|
|
def test_finally_rearms_after_an_exception(self):
|
|
p = FakePlug(fail_after_off=True)
|
|
try:
|
|
self.run_cut(p)
|
|
except Exception:
|
|
pass
|
|
self.assertTrue(p.on, 'finally: must switch the relay back on after an error')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|