1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
"""
Part of the bsfs test suite.
A copy of the license is provided with the project.
Author: Matthias Baumgartner, 2022
"""
# standard imports
import operator
import re
import unittest
# external imports
import rdflib
# bsie imports
from bsfs.namespace import ns
# objects to test
from bsfs.triple_store.sparql.utils import GenHopName, Query
## code ##
class TestGenHopName(unittest.TestCase):
def test_next(self):
# baseline
self.assertEqual(next(GenHopName(prefix='?foo', start=123)), '?foo123')
# respects prefix
self.assertEqual(next(GenHopName(prefix='?bar', start=123)), '?bar123')
# respects start
self.assertEqual(next(GenHopName(prefix='?foo', start=321)), '?foo321')
# counts up
cnt = GenHopName(prefix='?foo', start=998)
self.assertEqual(next(cnt), '?foo998')
self.assertEqual(next(cnt), '?foo999')
self.assertEqual(next(cnt), '?foo1000')
self.assertEqual(next(cnt), '?foo1001')
def test_essentials(self):
# can get the prefix
self.assertEqual(GenHopName(prefix='?foo', start=123).prefix, '?foo')
# can get the counter
self.assertEqual(GenHopName(prefix='?foo', start=123).curr, 122)
class TestQuery(unittest.TestCase):
def setUp(self):
self.root_type = 'http://bsfs.ai/schema/Entity'
self.root_head = '?root'
self.select = (('?head', 'name'), )
self.where = f'?root <{ns.bse.tag}> ?head'
def test_essentials(self):
# can access members
q = Query(self.root_type, self.root_head, self.select, self.where)
self.assertEqual(q.root_type, self.root_type)
self.assertEqual(q.root_head, self.root_head)
self.assertEqual(q.select, self.select)
self.assertEqual(q.where, self.where)
# comparison
self.assertEqual(q, Query(self.root_type, self.root_head, self.select, self.where))
self.assertEqual(hash(q), hash(Query(self.root_type, self.root_head, self.select, self.where)))
# comparison respects root_type
self.assertNotEqual(q, Query('http://bsfs.ai/schema/Tag', self.root_head, self.select, self.where))
self.assertNotEqual(hash(q), hash(Query('http://bsfs.ai/schema/Tag', self.root_head, self.select, self.where)))
# comparison respects root_head
self.assertNotEqual(q, Query(self.root_type, '?foo', self.select, self.where))
self.assertNotEqual(hash(q), hash(Query(self.root_type, '?foo', self.select, self.where)))
# comparison respects select
self.assertNotEqual(q, Query(self.root_type, self.root_head, (('?head', 'foo'), ), self.where))
self.assertNotEqual(hash(q), hash(Query(self.root_type, self.root_head, (('?head', 'foo'), ), self.where)))
# comparison respects where
self.assertNotEqual(q, Query(self.root_type, self.root_head, self.select, '?root bse:filename ?head'))
self.assertNotEqual(hash(q), hash(Query(self.root_type, self.root_head, self.select, '?root bse:filename ?head')))
# string conversion
self.assertEqual(str(q), q.query)
self.assertEqual(repr(q), "Query(http://bsfs.ai/schema/Entity, ?root, (('?head', 'name'),), ?root <http://bsfs.ai/schema/Entity#tag> ?head)")
def test_add(self):
q = Query(self.root_type, self.root_head, self.select, self.where)
# can only add a query
self.assertRaises(TypeError, operator.add, q, 1234)
self.assertRaises(TypeError, operator.add, q, 'foobar')
# root type and head must match
self.assertRaises(ValueError, operator.add, q, Query('http://bsfs.ai/schema/Tag', self.root_head))
self.assertRaises(ValueError, operator.add, q, Query(self.root_type, '?foobar'))
# select and were are combined
combo = q + Query(self.root_type, self.root_head, (('?foo', 'bar'), ), f'?root <{ns.bse.filename}> ?foo')
self.assertEqual(combo.select, (('?head', 'name'), ('?foo', 'bar')))
self.assertEqual(combo.where, f'?root <{ns.bse.tag}> ?head . ?root <{ns.bse.filename}> ?foo')
# select can be empty
combo = q + Query(self.root_type, self.root_head, None, f'?root <{ns.bse.filename}> ?foo')
self.assertEqual(combo.select, (('?head', 'name'), ))
combo = Query(self.root_type, self.root_head, None, f'?root <{ns.bse.filename}> ?foo') + q
self.assertEqual(combo.select, (('?head', 'name'), ))
combo = Query(self.root_type, self.root_head, None, self.where) + Query(self.root_type, self.root_head, None, f'?root <{ns.bse.filename}> ?foo')
self.assertEqual(combo.select, tuple())
# where can be empty
combo = q + Query(self.root_type, self.root_head, (('?foo', 'bar'), ))
self.assertEqual(combo.where, self.where)
combo = Query(self.root_type, self.root_head, (('?foo', 'bar'), )) + q
self.assertEqual(combo.where, self.where)
combo = Query(self.root_type, self.root_head, self.select) + Query(self.root_type, self.root_head, (('?foo', 'bar'), ))
self.assertEqual(combo.where, '')
def test_names(self):
self.assertEqual(Query(self.root_type, self.root_head, (('?head', 'name'), ), self.where).names,
('name', ))
self.assertEqual(Query(self.root_type, self.root_head, (('?head', 'name'), ('?foo', 'bar')), self.where).names,
('name', 'bar'))
def test_query(self):
def normalize(value):
value = value.strip()
value = value.lower()
value = value.replace(r'\n', ' ')
value, _ = re.subn('\s\s+', ' ', value)
return value
# query composes a valid query
q = Query(self.root_type, self.root_head, self.select, self.where)
self.assertEqual(normalize(q.query), normalize(f'select ?root (?head as ?name) where {{ ?root <{ns.rdf.type}>/<{ns.rdfs.subClassOf}>* <http://bsfs.ai/schema/Entity> . ?root <{ns.bse.tag}> ?head }}'))
# select and where are optional
q = Query(self.root_type, self.root_head)
self.assertEqual(normalize(q.query), normalize(f'select ?root where {{ ?root <{ns.rdf.type}>/<{ns.rdfs.subClassOf}>* <http://bsfs.ai/schema/Entity> . }}'))
# select and where need not to correspond
q = Query(self.root_type, self.root_head, (('?head', 'name'), ))
self.assertEqual(normalize(q.query), normalize(f'select ?root (?head as ?name) where {{ ?root <{ns.rdf.type}>/<{ns.rdfs.subClassOf}>* <http://bsfs.ai/schema/Entity> . }}'))
# query is used for string representation
self.assertEqual(str(q), q.query)
def test_call(self):
graph = rdflib.Graph()
# schema
graph.add((rdflib.URIRef('http://bsfs.ai/schema/Document'), rdflib.URIRef(ns.rdfs.subClassOf), rdflib.URIRef('http://bsfs.ai/schema/Entity')))
# nodes
graph.add((rdflib.URIRef('http://example.com/entity#1234'), rdflib.URIRef(ns.rdf.type), rdflib.URIRef('http://bsfs.ai/schema/Entity')))
graph.add((rdflib.URIRef('http://example.com/doc#1234'), rdflib.URIRef(ns.rdf.type), rdflib.URIRef('http://bsfs.ai/schema/Document')))
# links
graph.add((rdflib.URIRef('http://example.com/entity#1234'), rdflib.URIRef(ns.bse.tag), rdflib.Literal('tag#1234', datatype=rdflib.XSD.string)))
graph.add((rdflib.URIRef('http://example.com/doc#1234'), rdflib.URIRef(ns.bse.tag), rdflib.Literal('tag#1234', datatype=rdflib.XSD.string)))
# run query on a given graph
query = Query(self.root_type, self.root_head, self.select, self.where)
self.assertSetEqual(set(query(graph)), {
(rdflib.URIRef('http://example.com/entity#1234'), rdflib.Literal('tag#1234', datatype=rdflib.XSD.string)),
(rdflib.URIRef('http://example.com/doc#1234'), rdflib.Literal('tag#1234', datatype=rdflib.XSD.string)),
})
# query actually considers the passed graph
self.assertSetEqual(set(query(rdflib.Graph())), set())
## main ##
if __name__ == '__main__':
unittest.main()
## EOF ##
|
