light9 Changeset - 2cb47aa838ee

Changeset - 2cb47aa838ee

Parent rev.

Child rev.

[Not reviewed]

default

0 2 1

Drew Perttula - 9 years ago 2016-05-30 21:52:47
drewp@bigasterisk.com

fix collector_test using a new MockSyncedGraph
Ignore-this: b746b8d77427fab0a54376f3fa147a42

3 files changed with 84 insertions and 18 deletions:

light9/collector/collector_test.py

light9/collector/device_test.py

light9/rdfdb/mock_syncedgraph.py

0 comments (0 inline, 0 general)

light9/collector/collector_test.py

➞

Show inline comments

 import unittest
 import datetime
 from freezegun import freeze_time
 from rdflib import  Literal, Graph, Namespace
 from rdflib.parser import StringInputSource
 from rdflib import Namespace
 from light9.namespaces import L9, DEV
 from light9.collector.collector import Collector, outputMap
 from light9.rdfdb.mock_syncedgraph import MockSyncedGraph
 UDMX = Namespace('http://light9.bigasterisk.com/output/udmx/')
 DMX0 = Namespace('http://light9.bigasterisk.com/output/dmx0/')
 def fromN3(n3):
     out = Graph()
     out.parse(StringInputSource('''
         @prefix : <http://light9.bigasterisk.com/> .
 PREFIX = '''
    @prefix : <http://light9.bigasterisk.com/> .
         @prefix dev: <http://light9.bigasterisk.com/device/> .
         @prefix udmx: <http://light9.bigasterisk.com/output/udmx/> .
         @prefix dmx0: <http://light9.bigasterisk.com/output/dmx0/> .
     ''' + n3), format='n3')
     return out
 '''
 class MockOutput(object):
     def __init__(self, connections):
@@ @@ -38,27 +34,27 @@ class MockOutput(object): @@
 class TestOutputMap(unittest.TestCase):
     def testWorking(self):
         out0 = MockOutput([(0, DMX0['c1'])])
-        m = outputMap(fromN3('''
+        m = outputMap(MockSyncedGraph(PREFIX + '''
           dmx0:c1 :connectedTo dev:inst1Brightness .
           dev:inst1 a :Device; :brightness dev:inst1Brightness .
         '''), [out0])
         self.assertEqual({(DEV['inst1'], L9['brightness']): (out0, 0)}, m)
     def testMissingOutput(self):
         out0 = MockOutput([(0, DMX0['c1'])])
-        self.assertRaises(KeyError, outputMap, fromN3('''
+        self.assertRaises(KeyError, outputMap, MockSyncedGraph(PREFIX + '''
           dmx0:c2 :connectedTo dev:inst1Brightness .
           dev:inst1 a :Device; :brightness dev:inst1Brightness .
         '''), [out0])
     def testMissingOutputConnection(self):
         out0 = MockOutput([(0, DMX0['c1'])])
-        self.assertRaises(ValueError, outputMap, fromN3('''
+        self.assertRaises(ValueError, outputMap, MockSyncedGraph(PREFIX + '''
           dev:inst1 a :Device; :brightness dev:inst1Brightness .
         '''), [out0])
     def testMultipleOutputConnections(self):
         out0 = MockOutput([(0, DMX0['c1'])])
-        self.assertRaises(ValueError, outputMap, fromN3('''
+        self.assertRaises(ValueError, outputMap, MockSyncedGraph(PREFIX + '''
           dmx0:c1 :connectedTo dev:inst1Brightness .
           dmx0:c2 :connectedTo dev:inst1Brightness .
           dev:inst1 a :Device; :brightness dev:inst1Brightness .
@@ @@ -68,7 +64,7 @@ class TestOutputMap(unittest.TestCase): @@
 class TestCollector(unittest.TestCase):
     def setUp(self):
-        self.config = fromN3('''
+        self.config = MockSyncedGraph(PREFIX + '''
         udmx:c1 :connectedTo dev:colorStripRed .
         udmx:c2 :connectedTo dev:colorStripGreen .
@@ @@ -151,10 +147,29 @@ class TestCollector(unittest.TestCase): @@
         self.assertEqual([[204], 'flush', [127], 'flush'], self.dmx0.updates)
     def testNewSessionDropsPreviousSettingsOfOtherAttrs(self):
         c = Collector(self.config, outputs=[self.dmx0, self.udmx])
         c = Collector(MockSyncedGraph(PREFIX + '''
         udmx:c1 :connectedTo dev:colorStripRed .
         udmx:c2 :connectedTo dev:colorStripGreen .
         udmx:c3 :connectedTo dev:colorStripBlue .
         udmx:c4 :connectedTo dev:colorStripMode .
         c.setAttrs('client1', 'sess1', [(DEV['colorStrip'], L9['red'], 1)])
         c.setAttrs('client1', 'sess2', [(DEV['colorStrip'], L9['green'], 1)])
         dev:colorStrip a :Device, :ChauvetColorStrip;
           :red dev:colorStripRed;
           :green dev:colorStripGreen;
           :blue dev:colorStripBlue;
           :mode dev:colorStripMode .
         dmx0:c1 :connectedTo dev:inst1Brightness .
         dev:inst1 a :Device, :Dimmer;
           :brightness dev:inst1Brightness .
         '''), outputs=[self.dmx0, self.udmx])
         c.setAttrs('client1', 'sess1',
                    [(DEV['colorStrip'], L9['color'], '#ff0000')])
         c.setAttrs('client1', 'sess2',
                    [(DEV['colorStrip'], L9['color'], '#00ff00')])
         self.assertEqual([[255, 0, 0, 215], 'flush',
                           [0, 255, 0, 215], 'flush'], self.udmx.updates)

light9/collector/device_test.py

➞

Show inline comments

@@ @@ -48,5 +48,6 @@ class TestResolve(unittest.TestCase): @@
                          resolve(None, L9['color'], ['#ff0300', '#000400']))
     def testMaxes3Colors(self):
         self.assertEqual('#112233',
                          resolve(None, L9['color'], ['#110000', '#002200', '#000033']))
                          resolve(None, L9['color'],
                                  ['#110000', '#002200', '#000033']))

light9/rdfdb/mock_syncedgraph.py

➞

Show inline comments

@@ new file 100644 @@
 from rdflib import Graph, RDF, RDFS
 from rdflib.parser import StringInputSource
 class MockSyncedGraph(object):
     """
     Lets users of SyncedGraph mostly work. Doesn't yet help with any
     testing of the rerun-upon-graph-change behavior.
     """
     def __init__(self, n3Content):
         self._graph = Graph()
         self._graph.parse(StringInputSource(n3Content), format='n3')
     def addHandler(self, func):
         func()
     def value(self, subject=None, predicate=RDF.value, object=None,
               default=None, any=True):
         if object is not None:
             raise NotImplementedError()
         return self._graph.value(subject, predicate, object=object,
                                  default=default, any=any)
     def objects(self, subject=None, predicate=None):
         return self._graph.objects(subject, predicate)
     def label(self, uri):
         return self.value(uri, RDFS.label)
     def subjects(self, predicate=None, object=None):
         return self._graph.subjects(predicate, object)
     def predicate_objects(self, subject):
         return self._graph.predicate_objects(subject)
     def items(self, listUri):
         """generator. Having a chain of watchers on the results is not
         well-tested yet"""
         chain = set([listUri])
         while listUri:
             item = self.value(listUri, RDF.first)
             if item:
                 yield item
             listUri = self.value(listUri, RDF.rest)
             if listUri in chain:
                 raise ValueError("List contains a recursive rdf:rest reference")
             chain.add(listUri)
     def contains(self, triple):
         return triple in self._graph

0 comments (0 inline, 0 general)