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
|
import unittest
from tempfile import mktemp
from rdflib import ConjunctiveGraph, URIRef
from rdflib.store import VALID_STORE
from rdflib.plugins.stores.berkeleydb import has_bsddb
class BerkeleyDBTestCase(unittest.TestCase):
def setUp(self):
if not has_bsddb:
self.skipTest("skipping as berkleydb is missing")
self.store_name = "BerkeleyDB"
self.path = mktemp()
self.g = ConjunctiveGraph(store=self.store_name)
self.rt = self.g.open(self.path, create=True)
assert self.rt == VALID_STORE, "The underlying store is corrupt"
assert (
len(self.g) == 0
), "There must be zero triples in the graph just after store (file) creation"
data = """
PREFIX : <https://example.org/>
:a :b :c .
:d :e :f .
:d :g :h .
"""
self.g.parse(data=data, format="ttl")
def tearDown(self):
self.g.close()
def test_write(self):
assert (
len(self.g) == 3
), "There must be three triples in the graph after the first data chunk parse"
data2 = """
PREFIX : <https://example.org/>
:d :i :j .
"""
self.g.parse(data=data2, format="ttl")
assert (
len(self.g) == 4
), "There must be four triples in the graph after the second data chunk parse"
data3 = """
PREFIX : <https://example.org/>
:d :i :j .
"""
self.g.parse(data=data3, format="ttl")
assert (
len(self.g) == 4
), "There must still be four triples in the graph after the thrd data chunk parse"
def test_read(self):
sx = None
for s in self.g.subjects(
predicate=URIRef("https://example.org/e"),
object=URIRef("https://example.org/f"),
):
sx = s
assert sx == URIRef("https://example.org/d")
def test_sparql_query(self):
q = """
PREFIX : <https://example.org/>
SELECT (COUNT(*) AS ?c)
WHERE {
:d ?p ?o .
}"""
c = 0
for row in self.g.query(q):
c = int(row.c)
assert c == 2, "SPARQL COUNT must return 2"
def test_sparql_insert(self):
q = """
PREFIX : <https://example.org/>
INSERT DATA {
:x :y :z .
}"""
self.g.update(q)
assert len(self.g) == 4, "After extra triple insert, length must be 4"
def test_multigraph(self):
q = """
PREFIX : <https://example.org/>
INSERT DATA {
GRAPH :m {
:x :y :z .
}
GRAPH :n {
:x :y :z .
}
}"""
self.g.update(q)
q = """
SELECT (COUNT(?g) AS ?c)
WHERE {
SELECT DISTINCT ?g
WHERE {
GRAPH ?g {
?s ?p ?o
}
}
}
"""
c = 0
for row in self.g.query(q):
c = int(row.c)
assert c == 3, "SPARQL COUNT must return 3 (default, :m & :n)"
def test_open_shut(self):
assert len(self.g) == 3, "Initially we must have 3 triples from setUp"
self.g.close()
self.g = None
# reopen the graph
self.g = ConjunctiveGraph("BerkeleyDB")
self.g.open(self.path, create=False)
assert (
len(self.g) == 3
), "After close and reopen, we should still have the 3 originally added triples"
|
