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 32 insertions and 31 deletions:
light9/collector/collector_test.py
32
31
0 comments (0 inline, 0 general)
light9/collector/collector_test.py
➞
Show inline comments
 
@@ -3,6 +3,7 @@ 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

			




	
	
		
 
@@ -75,7 +76,7 @@ class MockWebListeners(WebListeners):

			




	
	
	
		
 
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;

			




	
	
		
 
@@ -93,9 +94,9 @@ class TestCollector(unittest.TestCase):

			




	
	
	
		
 
        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],

			




	
	
		
 
@@ -105,29 +106,29 @@ class TestCollector(unittest.TestCase):

			




	
	
	
		
 
        ], 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)

			




	
	
		
 
@@ -135,10 +136,10 @@ class TestCollector(unittest.TestCase):

			




	
	
	
		
 

			




	
	
	
		
 
    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)

			




	
	
	
		
 

			




	
	
		
 
@@ -159,45 +160,45 @@ class TestCollector(unittest.TestCase):

			




	
	
	
		
 
                      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