Added Glitch Room Source code to Gitty

tests/gpio/test_bridge.py

@@ -0,0 +1,105 @@
+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()