|
1630
|
1 import unittest
|
|
|
2
|
|
|
3 from rdflib import BNode, Literal, Namespace
|
|
|
4
|
|
|
5 from mqtt_message import graphFromMessage
|
|
|
6
|
|
|
7 ROOM = Namespace('http://projects.bigasterisk.com/room/')
|
|
|
8 JSON = Namespace('http://bigasterisk.com/anyJson/')
|
|
|
9
|
|
|
10
|
|
|
11 class TestGraphFromMessage(unittest.TestCase):
|
|
|
12
|
|
|
13 def testTopicOutput(self):
|
|
|
14 graph = graphFromMessage(b'a/b/topic', b'body')
|
|
|
15 self.assertEqual(len(graph), 9)
|
|
|
16
|
|
|
17 def testFloatBody(self):
|
|
|
18 graph = graphFromMessage(b'a/b/topic', b'3.3')
|
|
|
19 self.assertEqual(list(graph.objects(None, ROOM['bodyFloat'])), [Literal(3.3)])
|
|
|
20
|
|
|
21 def testStrBody(self):
|
|
|
22 graph = graphFromMessage(b'a/b/topic', b'3.x')
|
|
|
23 self.assertEqual(list(graph.objects(None, ROOM['body'])), [Literal("3.x")])
|
|
|
24
|
|
|
25 def testJsonEmptyBody(self):
|
|
|
26 graph = graphFromMessage(b'x', b'{}')
|
|
|
27 [jsonRoot] = graph.objects(None, ROOM['bodyJson'])
|
|
|
28 self.assertIsInstance(jsonRoot, BNode)
|
|
|
29
|
|
|
30 def testJsonBody(self):
|
|
|
31 graph = graphFromMessage(b'x', b'{"one":2}')
|
|
|
32 [jsonRoot] = graph.objects(None, ROOM['bodyJson'])
|
|
|
33 [(p, o)] = graph.predicate_objects(jsonRoot)
|
|
|
34 self.assertEqual(p, JSON['one'])
|
|
|
35 self.assertEqual(o, Literal(2))
|
