106 lines
4.2 KiB
Python
106 lines
4.2 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
import unittest
|
|
import tempfile
|
|
|
|
source = Path(__file__).resolve().parents[2] / "tools/pi_gpio/pi_gpio_bridge.py"
|
|
spec = importlib.util.spec_from_file_location("bridge", source)
|
|
bridge = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(bridge)
|
|
|
|
|
|
class FakePin:
|
|
def __init__(self, pin, active_high, initial_value):
|
|
self.pin, self.active_high, self.active = pin, active_high, initial_value
|
|
self.closed = False
|
|
self.activations = 0
|
|
|
|
def on(self):
|
|
self.active = True
|
|
self.activations += 1
|
|
def off(self): self.active = False
|
|
def close(self): self.closed = True
|
|
|
|
|
|
class BridgeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.now = 0
|
|
self.outputs = bridge.Outputs(FakePin, "cabinet_01_gpio", [17], lambda: self.now)
|
|
self.request = dict(controller="cabinet_01_gpio", pin=17, mode="pulse", active_high=True, duration_ms=600)
|
|
|
|
def test_pulse_and_retrigger(self):
|
|
self.outputs.command(self.request)
|
|
pin = self.outputs.pins[17]["device"]
|
|
self.assertTrue(pin.active)
|
|
self.now = .4
|
|
self.outputs.command(self.request)
|
|
self.now = .7
|
|
self.outputs.tick()
|
|
self.assertTrue(pin.active)
|
|
self.now = 1.01
|
|
self.outputs.tick()
|
|
self.assertFalse(pin.active)
|
|
|
|
def test_active_low_and_shutdown(self):
|
|
self.outputs.command(dict(self.request, active_high=False))
|
|
pin = self.outputs.pins[17]["device"]
|
|
self.assertFalse(pin.active_high)
|
|
self.assertTrue(pin.active)
|
|
self.outputs.close()
|
|
self.assertFalse(pin.active)
|
|
self.assertTrue(pin.closed)
|
|
|
|
def test_digital_off_and_watchdog(self):
|
|
self.outputs.command(dict(self.request, mode="set"))
|
|
self.now = 2
|
|
self.outputs.heartbeat()
|
|
self.now = 4
|
|
self.outputs.tick()
|
|
self.assertTrue(self.outputs.pins[17]["device"].active)
|
|
self.now = 5.1
|
|
self.outputs.tick()
|
|
self.assertFalse(self.outputs.pins[17]["device"].active)
|
|
self.outputs.command(dict(self.request, mode="set"))
|
|
self.outputs.command(dict(self.request, mode="set", active=False))
|
|
self.assertFalse(self.outputs.pins[17]["device"].active)
|
|
|
|
def test_reject_wrong_controller_pin_and_duration(self):
|
|
for changes in [dict(controller="cabinet_02_gpio"), dict(pin=18), dict(duration_ms=0), dict(duration_ms=10001), dict(active_high="false")]:
|
|
with self.assertRaises(ValueError):
|
|
self.outputs.command(dict(self.request, **changes))
|
|
self.assertFalse(self.outputs.pins)
|
|
|
|
def test_polarity_cannot_change_while_owned(self):
|
|
self.outputs.command(self.request)
|
|
with self.assertRaises(ValueError):
|
|
self.outputs.command(dict(self.request, active_high=False))
|
|
|
|
def test_reward_dedup_survives_restart(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
path = Path(temp) / "requests.sqlite3"
|
|
outputs = bridge.Outputs(FakePin, "cabinet_01_gpio", [17], ledger=path)
|
|
request = dict(self.request, request_id="reward-1")
|
|
self.assertTrue(outputs.command(request)["ok"])
|
|
self.assertTrue(outputs.command(request)["ok"])
|
|
self.assertEqual(outputs.pins[17]["device"].activations, 1)
|
|
outputs.close()
|
|
outputs = bridge.Outputs(FakePin, "cabinet_01_gpio", [17], ledger=path)
|
|
self.assertTrue(outputs.command(request)["ok"])
|
|
self.assertFalse(outputs.pins, "Duplicate must not even reopen output")
|
|
with self.assertRaises(ValueError):
|
|
outputs.command(dict(request, duration_ms=900))
|
|
outputs.close()
|
|
|
|
def test_incomplete_reward_is_not_replayed(self):
|
|
request = dict(self.request, request_id="uncertain")
|
|
import hashlib, json
|
|
digest = hashlib.sha256(json.dumps(request, sort_keys=True).encode()).hexdigest()
|
|
with self.outputs.ledger:
|
|
self.outputs.ledger.execute("INSERT INTO requests VALUES (?, ?, NULL)", ("uncertain", digest))
|
|
self.assertFalse(self.outputs.command(request)["ok"])
|
|
self.assertFalse(self.outputs.pins)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|