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
33
32
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

			




	
	
		
 
@@ -26,13 +27,13 @@ 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 ],

			




	
	
		
 
@@ -72,13 +73,13 @@ class MockWebListeners(WebListeners):

			




	
	
	
		
 
        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;

			




	
	
		
 
@@ -90,58 +91,58 @@ class TestCollector(unittest.TestCase):

			




	
	
	
		
 
        ''')

			




	
	
	
		
 

			




	
	
	
		
 
        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 + '''

			




	
	
	
		
 

			




	
	
		
 
@@ -156,48 +157,48 @@ class TestCollector(unittest.TestCase):

			




	
	
	
		
 
          :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)
    





    


  

  







  


    




    








    
        Server instance: kallithea-749ff669b6-v9g5f-75
    
    
        This site is powered by
            Kallithea 0.7.0,
        which is
        © 2010–2021 by various authors & licensed under GPLv3.
            – Support