light9 Changeset - ae38d21e6f6b

Changeset - ae38d21e6f6b

Parent rev.

Child rev.

[Not reviewed]

default

0 1 0

drewp@bigasterisk.com - 20 months ago 2023-05-22 07:56:19
drewp@bigasterisk.com

stricter types

1 file changed with 33 insertions and 32 deletions:

light9/collector/collector_test.py

0 comments (0 inline, 0 general)

light9/collector/collector_test.py

➞

Show inline comments

 import datetime
 import time
 import unittest
 from freezegun import freeze_time
 from light9.effect.settings import DeviceSettings
 from rdflib import Namespace
 from light9.collector.collector import Collector
 from light9.collector.output import Output
 from light9.collector.weblisteners import WebListeners
 from light9.mock_syncedgraph import MockSyncedGraph
 from light9.namespaces import DEV, L9
 from light9.newtypes import (ClientSessionType, ClientType, DeviceAttr, DeviceUri, HexColor, UnixTime)
 UDMX = Namespace('http://light9.bigasterisk.com/output/udmx/')
 DMX0 = Namespace('http://light9.bigasterisk.com/output/dmx0/')
@@ @@ -20,25 +21,25 @@ PREFIX = ''' @@
         @prefix dev: <http://light9.bigasterisk.com/device/> .
         @prefix udmx: <http://light9.bigasterisk.com/output/udmx/> .
         @prefix dmx0: <http://light9.bigasterisk.com/output/dmx0/> .
 '''
 THEATER = '''
         :brightness         a :DeviceAttr; :dataType :scalar .
         :SimpleDimmer a :DeviceClass;
           :deviceAttr :brightness;
           :attr
             [ :outputAttr :level; :dmxOffset 0 ] .
         :ChauvetColorStrip a :DeviceClass;
           :deviceAttr :color;
           :attr
             [ :outputAttr :mode;  :dmxOffset 0 ],
             [ :outputAttr :red;   :dmxOffset 1 ],
             [ :outputAttr :green; :dmxOffset 2 ],
             [ :outputAttr :blue;  :dmxOffset 3 ] .
 '''
 t0 = UnixTime(0)
 client1 = ClientType('client1')
@@ @@ -66,138 +67,138 @@ class MockOutput(Output): @@
 class MockWebListeners(WebListeners):
     def __init__(self):
         "do not init"
     def outputAttrsSet(self, *a, **kw):
         pass
 class TestCollector(unittest.TestCase):
     def setUp(self):
-        self.config = MockSyncedGraph(PREFIX + THEATER + '''
+        self.graph = MockSyncedGraph(PREFIX + THEATER + '''
         dev:colorStrip a :Device, :ChauvetColorStrip;
           :dmxUniverse udmx:; :dmxBase 1;
           :red dev:colorStripRed;
           :green dev:colorStripGreen;
           :blue dev:colorStripBlue;
           :mode dev:colorStripMode .
         dev:inst1 a :Device, :SimpleDimmer;
           :dmxUniverse dmx0:; :dmxBase 1;
           :level dev:inst1Brightness .
         ''')
         self.dmx0 = MockOutput(DMX0, [(0, DMX0['c1'])])
         self.udmx = MockOutput(UDMX, [(0, UDMX['c1']), (1, UDMX['c2']), (2, UDMX['c3']), (3, UDMX['c4'])])
     def testRoutesColorOutput(self):
-        c = Collector(self.config, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
+        c = Collector(self.graph, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
         c.setAttrs(client1, session1, [(colorStrip, color, HexColor('#00ff00'))], t0)
+        c.setAttrs(client1, session1, DeviceSettings(self.graph, [(colorStrip, color, HexColor('#00ff00'))]), t0)
         self.assertEqual([
             [215, 0, 255, 0],
         ], self.udmx.updates)
         self.assertEqual([
             [0, 0, 0, 0],
         ], self.dmx0.updates)
     def testOutputMaxOfTwoClients(self):
-        c = Collector(self.config, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
+        c = Collector(self.graph, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
         c.setAttrs(client1, session1, [(colorStrip, color, HexColor('#ff0000'))], t0)
         c.setAttrs(client2, session1, [(colorStrip, color, HexColor('#333333'))], t0)
         c.setAttrs(client1, session1, DeviceSettings(self.graph, [(colorStrip, color, HexColor('#ff0000'))]), t0)
         c.setAttrs(client2, session1, DeviceSettings(self.graph, [(colorStrip, color, HexColor('#333333'))]), t0)
         self.assertEqual([[215, 255, 0, 0], [215, 255, 51, 51]], self.udmx.updates)
         self.assertEqual([[0, 0, 0, 0], [0, 0, 0, 0]], self.dmx0.updates)
     def testClientOnSameOutputIsRememberedOverCalls(self):
-        c = Collector(self.config, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
+        c = Collector(self.graph, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
         c.setAttrs(client1, session1, [(colorStrip, color, HexColor('#080000'))], t0)
         c.setAttrs(client2, session1, [(colorStrip, color, HexColor('#060000'))], t0)
         c.setAttrs(client1, session1, [(colorStrip, color, HexColor('#050000'))], t0)
         c.setAttrs(client1, session1, DeviceSettings(self.graph, [(colorStrip, color, HexColor('#080000'))]), t0)
         c.setAttrs(client2, session1, DeviceSettings(self.graph, [(colorStrip, color, HexColor('#060000'))]), t0)
         c.setAttrs(client1, session1, DeviceSettings(self.graph, [(colorStrip, color, HexColor('#050000'))]), t0)
         self.assertEqual([[215, 8, 0, 0], [215, 8, 0, 0], [215, 6, 0, 0]], self.udmx.updates)
         self.assertEqual([[0, 0, 0, 0], [0, 0, 0, 0], [0, 0, 0, 0]], self.dmx0.updates)
     def testClientsOnDifferentOutputs(self):
-        c = Collector(self.config, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
+        c = Collector(self.graph, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
         c.setAttrs(client1, session1, [(colorStrip, color, HexColor('#aa0000'))], t0)
         c.setAttrs(client2, session1, [(inst1, brightness, .5)], t0)
         c.setAttrs(client1, session1, DeviceSettings(self.graph, [(colorStrip, color, HexColor('#aa0000'))]), t0)
         c.setAttrs(client2, session1, DeviceSettings(self.graph, [(inst1, brightness, .5)]), t0)
         # ok that udmx is flushed twice- it can screen out its own duplicates
         self.assertEqual([[215, 170, 0, 0], [215, 170, 0, 0]], self.udmx.updates)
         self.assertEqual([[0, 0, 0, 0], [127, 0, 0, 0]], self.dmx0.updates)
     def testNewSessionReplacesPreviousOutput(self):
         # ..as opposed to getting max'd with it
-        c = Collector(self.config, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
+        c = Collector(self.graph, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
         c.setAttrs(client1, session1, [(inst1, brightness, .8)], t0)
         c.setAttrs(client1, session2, [(inst1, brightness, .5)], t0)
         c.setAttrs(client1, session1, DeviceSettings(self.graph, [(inst1, brightness, .8)]), t0)
         c.setAttrs(client1, session2, DeviceSettings(self.graph, [(inst1, brightness, .5)]), t0)
         self.assertEqual([[204, 0, 0, 0], [127, 0, 0, 0]], self.dmx0.updates)
     def testNewSessionDropsPreviousSettingsOfOtherAttrs(self):
         c = Collector(MockSyncedGraph(PREFIX + THEATER + '''
         dev:colorStrip a :Device, :ChauvetColorStrip;
           :dmxUniverse udmx:; :dmxBase 1;
           :red dev:colorStripRed;
           :green dev:colorStripGreen;
           :blue dev:colorStripBlue;
           :mode dev:colorStripMode .
         dev:inst1 a :Device, :SimpleDimmer;
           :dmxUniverse dmx0:; :dmxBase 0;
           :level dev:inst1Brightness .
         '''),
                       outputs=[self.dmx0, self.udmx],
                       listeners=MockWebListeners())
         c.setAttrs(client1, session1, [(colorStrip, color, HexColor('#ff0000'))], t0)
         c.setAttrs(client1, session2, [(colorStrip, color, HexColor('#00ff00'))], t0)
         c.setAttrs(client1, session1, DeviceSettings(self.graph, [(colorStrip, color, HexColor('#ff0000'))]), t0)
         c.setAttrs(client1, session2, DeviceSettings(self.graph, [(colorStrip, color, HexColor('#00ff00'))]), t0)
         self.assertEqual([[215, 255, 0, 0], [215, 0, 255, 0]], self.udmx.updates)
     def testClientIsForgottenAfterAWhile(self):
         with freeze_time(datetime.datetime.now()) as ft:
             c = Collector(self.config, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
             c.setAttrs(client1, session1, [(inst1, brightness, .5)], UnixTime(time.time()))
             c = Collector(self.graph, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
             c.setAttrs(client1, session1, DeviceSettings(self.graph, [(inst1, brightness, .5)]), UnixTime(time.time()))
             ft.tick(delta=datetime.timedelta(seconds=1))
             # this max's with client1's value so we still see .5
             c.setAttrs(client2, session1, [(inst1, brightness, .2)], UnixTime(time.time()))
+            c.setAttrs(client2, session1, DeviceSettings(self.graph, [(inst1, brightness, .2)]), UnixTime(time.time()))
             ft.tick(delta=datetime.timedelta(seconds=9.1))
             # now client1 is forgotten, so our value appears
             c.setAttrs(client2, session1, [(inst1, brightness, .4)], UnixTime(time.time()))
+            c.setAttrs(client2, session1, DeviceSettings(self.graph, [(inst1, brightness, .4)]), UnixTime(time.time()))
             self.assertEqual([[127, 0, 0, 0], [127, 0, 0, 0], [102, 0, 0, 0]], self.dmx0.updates)
     def testClientUpdatesAreNotMerged(self):
         # second call to setAttrs forgets the first
-        c = Collector(self.config, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
+        c = Collector(self.graph, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
         t0 = UnixTime(time.time())
         c.setAttrs(client1, session1, [(inst1, brightness, .5)], t0)
         c.setAttrs(client1, session1, [(inst1, brightness, 1)], t0)
         c.setAttrs(client1, session1, [(colorStrip, color, HexColor('#00ff00'))], t0)
         c.setAttrs(client1, session1, DeviceSettings(self.graph, [(inst1, brightness, .5)]), t0)
         c.setAttrs(client1, session1, DeviceSettings(self.graph, [(inst1, brightness, 1)]), t0)
         c.setAttrs(client1, session1, DeviceSettings(self.graph, [(colorStrip, color, HexColor('#00ff00'))]), t0)
         self.assertEqual([[215, 0, 0, 0], [215, 0, 0, 0], [215, 0, 255, 0]], self.udmx.updates)
         self.assertEqual([[127, 0, 0, 0], [255, 0, 0, 0], [0, 0, 0, 0]], self.dmx0.updates)
     def testRepeatedAttributesInOneRequestGetResolved(self):
-        c = Collector(self.config, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
+        c = Collector(self.graph, outputs=[self.dmx0, self.udmx], listeners=MockWebListeners())
         c.setAttrs(client1, session1, [
+        c.setAttrs(client1, session1, DeviceSettings(self.graph, [
             (inst1, brightness, .5),
             (inst1, brightness, .3),
         ], t0)
+        ]), t0)
         self.assertEqual([[127, 0, 0, 0]], self.dmx0.updates)
         c.setAttrs(client1, session1, [
+        c.setAttrs(client1, session1, DeviceSettings(self.graph, [
             (inst1, brightness, .3),
             (inst1, brightness, .5),
         ], t0)
+        ]), t0)
         self.assertEqual([[127, 0, 0, 0], [127, 0, 0, 0]], self.dmx0.updates)

0 comments (0 inline, 0 general)